Java中的synchronized关键字(二)
2019-03-03 本文已影响0人
buzzerrookie
上一篇文章简要分析了synchronized关键字，本文分析重量级监视器的实现以及如何获得监视器。
监视器的实现
Java的监视器在Hotspot虚拟机中由ObjectMonitor实现，ObjectMonitor类在文件hotspot/src/share/vm/runtime/objectMonitor.hpp中定义，其部分代码如下所示：
class ObjectMonitor {
public:
enum {
OM_OK, // no error
OM_SYSTEM_ERROR, // operating system error
OM_ILLEGAL_MONITOR_STATE, // IllegalMonitorStateException
OM_INTERRUPTED, // Thread.interrupt()
OM_TIMED_OUT // Object.wait() timed out
};
//省略一些代码
// initialize the monitor, exception the semaphore, all other fields
// are simple integers or pointers
ObjectMonitor() {
_header = NULL;
_count = 0;
_waiters = 0,
_recursions = 0;
_object = NULL;
_owner = NULL;
_WaitSet = NULL;
_WaitSetLock = 0 ;
_Responsible = NULL ;
_succ = NULL ;
_cxq = NULL ;
FreeNext = NULL ;
_EntryList = NULL ;
_SpinFreq = 0 ;
_SpinClock = 0 ;
OwnerIsThread = 0 ;
_previous_owner_tid = 0;
}
bool try_enter (TRAPS) ;
void enter(TRAPS);
void exit(bool not_suspended, TRAPS);
void wait(jlong millis, bool interruptable, TRAPS);
void notify(TRAPS);
void notifyAll(TRAPS);
//省略一些代码
private:
friend class ObjectSynchronizer;
friend class ObjectWaiter;
friend class VMStructs;
// WARNING: this must be the very first word of ObjectMonitor
// This means this class can't use any virtual member functions.
volatile markOop _header; // displaced object header word - mark
void* volatile _object; // backward object pointer - strong root
double SharingPad [1] ; // temp to reduce false sharing
// All the following fields must be machine word aligned
// The VM assumes write ordering wrt these fields, which can be
// read from other threads.
protected: // protected for jvmtiRawMonitor
void * volatile _owner; // pointer to owning thread OR BasicLock
volatile jlong _previous_owner_tid; // thread id of the previous owner of the monitor
volatile intptr_t _recursions; // recursion count, 0 for first entry
private:
int OwnerIsThread ; // _owner is (Thread *) vs SP/BasicLock
ObjectWaiter * volatile _cxq ; // LL of recently-arrived threads blocked on entry.
// The list is actually composed of WaitNodes, acting
// as proxies for Threads.
protected:
ObjectWaiter * volatile _EntryList ; // Threads blocked on entry or reentry.
private:
Thread * volatile _succ ; // Heir presumptive thread - used for futile wakeup throttling
Thread * volatile _Responsible ;
int _PromptDrain ; // rqst to drain cxq into EntryList ASAP
volatile int _Spinner ; // for exit->spinner handoff optimization
volatile int _SpinFreq ; // Spin 1-out-of-N attempts: success rate
volatile int _SpinClock ;
volatile int _SpinDuration ;
volatile intptr_t _SpinState ; // MCS/CLH list of spinners
// TODO-FIXME: _count, _waiters and _recursions should be of
// type int, or int32_t but not intptr_t. There's no reason
// to use 64-bit fields for these variables on a 64-bit JVM.
volatile intptr_t _count; // reference count to prevent reclaimation/deflation
// at stop-the-world time. See deflate_idle_monitors().
// _count is approximately |_WaitSet| + |_EntryList|
protected:
volatile intptr_t _waiters; // number of waiting threads
private:
protected:
ObjectWaiter * volatile _WaitSet; // LL of threads wait()ing on the monitor
private:
volatile int _WaitSetLock; // protects Wait Queue - simple spinlock
//省略一些代码
}
ObjectMonitor类重要的成员变量和函数如下：
- _header：监视器所属对象的mark word；
- _object：监视器所属的对象；
- _owner：指向锁记录或持有监视器的线程；
- OwnerIsThread：_owner变量是线程指针时为1，是锁记录指针时是0；
- _EntryList：双向链表，保存在入口或重入阻塞的线程；
- _WaitSet：双向链表，保存等待的线程；
- _cxq：单向链表，保存在入口阻塞的最近到达的线程（Recently Arrived Threads，RATs）；
- 构造函数只是为成员变量赋初值，指针变量均被赋值为NULL，整型变量均被赋值为0；
- try_enter、enter、exit、wait、notify和notifyAll等函数均与获取和释放监视器有关。
要点说明
文件objectMonitor.cpp中的注释很有用，有助于理解监视器：
- 线程通过成功执行将_owner从NULL变为非NULL的CAS操作以获得监视器；
- 线程在某个时刻最多只出现在一个链表中，要么是cxq，要么是EntryList，要么是WaitSet；
- 竞争线程使用CAS将它们自己推进cxq，然后自旋或者park；
- 竞争线程最后获得锁之后必须将自己从EntryList或者cxq出队；
- 退出线程在EntryList上识别一个“法定继承人”并unpark，注意退出线程并没有将这个后继线程从EntryList解除链接，在unpark后，被唤醒的线程会重新竞争监视器的所有权，它要么获得锁要么重新park自己。退出线程不保证将所有权传递给后继线程（即这不是手递手式的），如果EntryList为空但cxq却不空，那么退出线程会通过分离cxq（利用CAS将cxq链表置为NULL）并将cxq上的元素转到EntryList。
- 只有监视器所有者才能访问或者更改EntryList；
- 只有监视器所有者才能分离cxq；
- notify()或notifyAll()只是将WaitSet中的线程移到EntryList或者cxq。
CAS操作
在分析与监视器有关的函数时，理解CAS操作至关重要。Hotspot虚拟机中的CAS操作由Atomic类的cmpxchg和cmpxchg_ptr静态函数实现，一般地，CAS操作均由处理器的特定指令支持。以x86上的Linux系统为例，cmpxchg和cmpxchg_ptr函数定义在文件hotspot/src/os_cpu/linux_x86/vm/atomic_linux_x86.inline.hpp里，代码如下：
inline jlong Atomic::cmpxchg (jlong exchange_value, volatile jlong* dest, jlong compare_value) {
bool mp = os::is_MP();
__asm__ __volatile__ (LOCK_IF_MP(%4) "cmpxchgq %1,(%3)"
: "=a" (exchange_value)
: "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
: "cc", "memory");
return exchange_value;
}
inline void* Atomic::cmpxchg_ptr(void* exchange_value, volatile void* dest, void* compare_value) {
return (void*)cmpxchg((jlong)exchange_value, (volatile jlong*)dest, (jlong)compare_value);
}
根据GCC内联汇编语法，cmpxchg函数的含义是如果compare_value与dest地址引用的内容相等，那么返回compare_value同时exchange_value被写入dest地址；否则返回dest地址引用的内容。
获得监视器
enter函数
ObjectMonitor类的enter函数用于获得监视器，其代码如下所示：
void ATTR ObjectMonitor::enter(TRAPS) { // TRAPS宏展开后是 Thread* __the_thread__
// The following code is ordered to check the most common cases first
// and to reduce RTS->RTO cache line upgrades on SPARC and IA32 processors.
Thread * const Self = THREAD ; // THREAD宏展开后是 __the_thread__
void * cur ;
cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
if (cur == NULL) { // CAS操作成功，说明CAS操作前_owner是NULL，监视器未被任何线程持有，CAS操作成功后_owner的值是参数线程
// Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
assert (_recursions == 0 , "invariant") ;
assert (_owner == Self, "invariant") ;
// CONSIDER: set or assert OwnerIsThread == 1
return ;
}
if (cur == Self) { // CAS操作失败的一种情况，CAS操作前_owner就已经是参数线程，这是重入了
// TODO-FIXME: check for integer overflow! BUGID 6557169.
_recursions ++ ;
return ;
}
// CAS操作失败的另一种情况，CAS操作前_owner是指向某个锁记录的指针，那么看这个锁记录是否由参数线程在它的栈桢上分配的，如果是则说明是该线程已经持有监视器
if (Self->is_lock_owned ((address)cur)) {
assert (_recursions == 0, "internal state error");
_recursions = 1 ;
// Commute owner from a thread-specific on-stack BasicLockObject address to
// a full-fledged "Thread *".
_owner = Self ;
OwnerIsThread = 1 ; // 当前_owner是线程指针
return ;
}
// CAS操作失败的其他情况，真正的锁竞争
// We've encountered genuine contention.
assert (Self->_Stalled == 0, "invariant") ;
Self->_Stalled = intptr_t(this) ;
// Try one round of spinning *before* enqueueing Self
// and before going through the awkward and expensive state
// transitions. The following spin is strictly optional ...
// Note that if we acquire the monitor from an initial spin
// we forgo posting JVMTI events and firing DTRACE probes.
if (Knob_SpinEarly && TrySpin (Self) > 0) { // 先尝试一轮自旋，以避免昂贵的状态转移操作，若TrySpin返回值大于0则说明在自旋的时候获得了监视器锁
assert (_owner == Self , "invariant") ;
assert (_recursions == 0 , "invariant") ;
assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
Self->_Stalled = 0 ;
return ;
}
assert (_owner != Self , "invariant") ;
assert (_succ != Self , "invariant") ;
assert (Self->is_Java_thread() , "invariant") ;
JavaThread * jt = (JavaThread *) Self ;
assert (!SafepointSynchronize::is_at_safepoint(), "invariant") ;
assert (jt->thread_state() != _thread_blocked , "invariant") ;
assert (this->object() != NULL , "invariant") ;
assert (_count >= 0, "invariant") ;
// Prevent deflation at STW-time. See deflate_idle_monitors() and is_busy().
// Ensure the object-monitor relationship remains stable while there's contention.
Atomic::inc_ptr(&_count);
EventJavaMonitorEnter event;
{ // Change java thread status to indicate blocked on monitor enter.
JavaThreadBlockedOnMonitorEnterState jtbmes(jt, this);
Self->set_current_pending_monitor(this);
DTRACE_MONITOR_PROBE(contended__enter, this, object(), jt);
if (JvmtiExport::should_post_monitor_contended_enter()) {
JvmtiExport::post_monitor_contended_enter(jt, this);
// The current thread does not yet own the monitor and does not
// yet appear on any queues that would get it made the successor.
// This means that the JVMTI_EVENT_MONITOR_CONTENDED_ENTER event
// handler cannot accidentally consume an unpark() meant for the
// ParkEvent associated with this ObjectMonitor.
}
OSThreadContendState osts(Self->osthread());
ThreadBlockInVM tbivm(jt);
// TODO-FIXME: change the following for(;;) loop to straight-line code.
for (;;) {
jt->set_suspend_equivalent();
// cleared by handle_special_suspend_equivalent_condition()
// or java_suspend_self()
EnterI (THREAD) ;
if (!ExitSuspendEquivalent(jt)) break ;
//
// We have acquired the contended monitor, but while we were
// waiting another thread suspended us. We don't want to enter
// the monitor while suspended because that would surprise the
// thread that suspended us.
//
_recursions = 0 ;
_succ = NULL ;
exit (false, Self) ;
jt->java_suspend_self();
}
Self->set_current_pending_monitor(NULL);
// We cleared the pending monitor info since we've just gotten past
// the enter-check-for-suspend dance and we now own the monitor free
// and clear, i.e., it is no longer pending. The ThreadBlockInVM
// destructor can go to a safepoint at the end of this block. If we
// do a thread dump during that safepoint, then this thread will show
// as having "-locked" the monitor, but the OS and java.lang.Thread
// states will still report that the thread is blocked trying to
// acquire it.
}
Atomic::dec_ptr(&_count);
assert (_count >= 0, "invariant") ;
Self->_Stalled = 0 ;
// Must either set _recursions = 0 or ASSERT _recursions == 0.
assert (_recursions == 0 , "invariant") ;
assert (_owner == Self , "invariant") ;
assert (_succ != Self , "invariant") ;
assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
// 省略一些代码
}
- 首先进行一次CAS操作，交换值是enter函数参数线程指针，目的地址是监视器的_owner字段地址，比较值是NULL。如果CAS操作成功，则说明CAS操作前_owner是NULL，监视器未被任何线程持有，CAS操作成功后_owner的值是参数线程（即上文要点说明的第一点）；
- 接着处理CAS操作失败的一种情况，CAS操作前_owner就已经是参数线程，这是重入了；
- 然后处理CAS操作失败的另一种情况，CAS操作前_owner是指向某个锁记录的指针，那么看这个锁记录是否由参数线程在它的栈桢上分配，如果是则说明是该线程持有这个监视器；
- 最后是CAS操作失败的其他情况，真正的锁竞争。先尝试一轮自旋，以避免昂贵的状态转移操作，若TrySpin返回值大于0则说明在自旋的时候获得了监视器，否则执行EnterI函数竞争监视器。
EnterI函数
EnterI函数的代码如下，第二个for循环印证了上文要点说明的第三点。
void ATTR ObjectMonitor::EnterI (TRAPS) {
Thread * Self = THREAD ;
assert (Self->is_Java_thread(), "invariant") ;
assert (((JavaThread *) Self)->thread_state() == _thread_blocked , "invariant") ;
// Try the lock - TATAS
if (TryLock (Self) > 0) { // 先尝试一下
assert (_succ != Self , "invariant") ;
assert (_owner == Self , "invariant") ;
assert (_Responsible != Self , "invariant") ;
return ;
}
DeferredInitialize () ;
// We try one round of spinning *before* enqueueing Self.
//
// If the _owner is ready but OFFPROC we could use a YieldTo()
// operation to donate the remainder of this thread's quantum
// to the owner. This has subtle but beneficial affinity
// effects.
if (TrySpin (Self) > 0) { // 入队之前再自旋试一下
assert (_owner == Self , "invariant") ;
assert (_succ != Self , "invariant") ;
assert (_Responsible != Self , "invariant") ;
return ;
}
// The Spin failed -- Enqueue and park the thread ...
assert (_succ != Self , "invariant") ;
assert (_owner != Self , "invariant") ;
assert (_Responsible != Self , "invariant") ;
// Enqueue "Self" on ObjectMonitor's _cxq.
//
// Node acts as a proxy for Self.
// As an aside, if were to ever rewrite the synchronization code mostly
// in Java, WaitNodes, ObjectMonitors, and Events would become 1st-class
// Java objects. This would avoid awkward lifecycle and liveness issues,
// as well as eliminate a subset of ABA issues.
// TODO: eliminate ObjectWaiter and enqueue either Threads or Events.
//
// 自旋失败，将参数线程包装成ObjectWaiter并加入cxq队首
ObjectWaiter node(Self) ;
Self->_ParkEvent->reset() ;
node._prev = (ObjectWaiter *) 0xBAD ; // 这个地址很魔幻，BAD :)
node.TState = ObjectWaiter::TS_CXQ ;
// Push "Self" onto the front of the _cxq.
// Once on cxq/EntryList, Self stays on-queue until it acquires the lock.
// Note that spinning tends to reduce the rate at which threads
// enqueue and dequeue on EntryList|cxq.
ObjectWaiter * nxt ;
for (;;) { // 更新链表头_cxq为node的地址，用CAS是因为可能有多个线程同时竞争监视器，都在运行这段代码，如果CAS失败则重试
node._next = nxt = _cxq ;
if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
// Interference - the CAS failed because _cxq changed. Just retry.
// As an optional optimization we retry the lock.
if (TryLock (Self) > 0) {
assert (_succ != Self , "invariant") ;
assert (_owner == Self , "invariant") ;
assert (_Responsible != Self , "invariant") ;
return ;
}
}
// 省略一些代码
for (;;) {
if (TryLock (Self) > 0) break ;
assert (_owner != Self, "invariant") ;
if ((SyncFlags & 2) && _Responsible == NULL) {
Atomic::cmpxchg_ptr (Self, &_Responsible, NULL) ;
}
// park self
if (_Responsible == Self || (SyncFlags & 1)) {
TEVENT (Inflated enter - park TIMED) ;
Self->_ParkEvent->park ((jlong) RecheckInterval) ;
// Increase the RecheckInterval, but clamp the value.
RecheckInterval *= 8 ;
if (RecheckInterval > 1000) RecheckInterval = 1000 ;
} else {
TEVENT (Inflated enter - park UNTIMED) ;
Self->_ParkEvent->park() ;
}
if (TryLock(Self) > 0) break ;
// The lock is still contested.
// Keep a tally of the # of futile wakeups.
// Note that the counter is not protected by a lock or updated by atomics.
// That is by design - we trade "lossy" counters which are exposed to
// races during updates for a lower probe effect.
TEVENT (Inflated enter - Futile wakeup) ;
if (ObjectMonitor::_sync_FutileWakeups != NULL) {
ObjectMonitor::_sync_FutileWakeups->inc() ;
}
++ nWakeups ;
// Assuming this is not a spurious wakeup we'll normally find _succ == Self.
// We can defer clearing _succ until after the spin completes
// TrySpin() must tolerate being called with _succ == Self.
// Try yet another round of adaptive spinning.
if ((Knob_SpinAfterFutile & 1) && TrySpin (Self) > 0) break ;
// We can find that we were unpark()ed and redesignated _succ while
// we were spinning. That's harmless. If we iterate and call park(),
// park() will consume the event and return immediately and we'll
// just spin again. This pattern can repeat, leaving _succ to simply
// spin on a CPU. Enable Knob_ResetEvent to clear pending unparks().
// Alternately, we can sample fired() here, and if set, forgo spinning
// in the next iteration.
if ((Knob_ResetEvent & 1) && Self->_ParkEvent->fired()) {
Self->_ParkEvent->reset() ;
OrderAccess::fence() ;
}
if (_succ == Self) _succ = NULL ;
// Invariant: after clearing _succ a thread *must* retry _owner before parking.
OrderAccess::fence() ;
}
// 上面的循环里参数线程park后会被阻塞，跳出循环后一定是持有了该监视器
// Egress :
// Self has acquired the lock -- Unlink Self from the cxq or EntryList.
// Normally we'll find Self on the EntryList .
// From the perspective of the lock owner (this thread), the
// EntryList is stable and cxq is prepend-only.
// The head of cxq is volatile but the interior is stable.
// In addition, Self.TState is stable.
assert (_owner == Self , "invariant") ;
assert (object() != NULL , "invariant") ;
// I'd like to write:
// guarantee (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
// but as we're at a safepoint that's not safe.
UnlinkAfterAcquire (Self, &node) ; // Self线程获得监视器后，将自己从EntryList链表或者cxq链表删除
if (_succ == Self) _succ = NULL ;
//省略一些代码
return ;
}
- 首先调用TryLock函数尝试获取锁，若失败则再自旋试一下；
- 若自旋失败，则将参数线程包装成ObjectWaiter并加入cxq队首；
- 更新链表头_cxq为上一步ObjectWaiter的地址，用CAS是因为可能有多个线程同时竞争监视器，都在运行这段代码，如果CAS失败则重试；
- 在第二个for循环里，利用TryLock函数不断尝试获得监视器，若失败则调用ParkEvent类的park函数将自己阻塞（在文章Java线程的中断与休眠中分析过park函数）；若成功则跳出循环，此时一定是持有了监视器；
- 参数线程获得监视器后，调用UnlinkAfterAcquire函数将自己从EntryList链表或者cxq链表删除（即上文要点说明的第四点）。
TryLock函数
TryLock函数功能较为简单，其代码如下：
// Caveat: TryLock() is not necessarily serializing if it returns failure.
// Callers must compensate as needed.
int ObjectMonitor::TryLock (Thread * Self) {
for (;;) {
void * own = _owner ;
if (own != NULL) return 0 ; // _owner不为NULL，说明已经有其他线程持有该监视器了
if (Atomic::cmpxchg_ptr (Self, &_owner, NULL) == NULL) { // CAS操作成功，说明CAS操作前_owner是NULL，监视器未被任何线程持有，CAS操作后_owner的值是参数线程
// Either guarantee _recursions == 0 or set _recursions = 0.
assert (_recursions == 0, "invariant") ;
assert (_owner == Self, "invariant") ;
// CONSIDER: set or assert that OwnerIsThread == 1
return 1 ;
}
// The lock had been free momentarily, but we lost the race to the lock.
// Interference -- the CAS failed.
// We can either return -1 or retry.
// Retry doesn't make as much sense because the lock was just acquired.
if (true) return -1 ;
}
}
- 若监视器已被其他线程占有则返回0；
- 若CAS操作成功地使参数线程持有该监视器，那么返回1；
- 若CAS操作失败，则其他线程已持有该监视器，返回-1。
UnlinkAfterAcquire函数
UnlinkAfterAcquire函数代码如下，作用是当参数线程获得监视器后将自己从EntryList双向链表或者cxq单向链表删除。
// after the thread acquires the lock in ::enter(). Equally, we could defer
// unlinking the thread until ::exit()-time.
void ObjectMonitor::UnlinkAfterAcquire (Thread * Self, ObjectWaiter * SelfNode)
{
assert (_owner == Self, "invariant") ;
assert (SelfNode->_thread == Self, "invariant") ;
if (SelfNode->TState == ObjectWaiter::TS_ENTER) { // 在EntryList链表
// Normal case: remove Self from the DLL EntryList .
// This is a constant-time operation.
ObjectWaiter * nxt = SelfNode->_next ;
ObjectWaiter * prv = SelfNode->_prev ;
if (nxt != NULL) nxt->_prev = prv ;
if (prv != NULL) prv->_next = nxt ;
if (SelfNode == _EntryList ) _EntryList = nxt ;
assert (nxt == NULL || nxt->TState == ObjectWaiter::TS_ENTER, "invariant") ;
assert (prv == NULL || prv->TState == ObjectWaiter::TS_ENTER, "invariant") ;
TEVENT (Unlink from EntryList) ;
} else { // 在cxq链表
guarantee (SelfNode->TState == ObjectWaiter::TS_CXQ, "invariant") ;
// Inopportune interleaving -- Self is still on the cxq.
// This usually means the enqueue of self raced an exiting thread.
// Normally we'll find Self near the front of the cxq, so
// dequeueing is typically fast. If needbe we can accelerate
// this with some MCS/CHL-like bidirectional list hints and advisory
// back-links so dequeueing from the interior will normally operate
// in constant-time.
// Dequeue Self from either the head (with CAS) or from the interior
// with a linear-time scan and normal non-atomic memory operations.
// CONSIDER: if Self is on the cxq then simply drain cxq into EntryList
// and then unlink Self from EntryList. We have to drain eventually,
// so it might as well be now.
ObjectWaiter * v = _cxq ; // 之前在EnterI函数中把Self加入cxq队列时是加到了队首，但此时有可能cxq队首已经改变
assert (v != NULL, "invariant") ;
if (v != SelfNode || Atomic::cmpxchg_ptr (SelfNode->_next, &_cxq, v) != v) {
// The CAS above can fail from interference IFF a "RAT" arrived.
// In that case Self must be in the interior and can no longer be
// at the head of cxq.
if (v == SelfNode) { // 这个if是为了处理以下情况：上面if之前v == SelfNode，那么上面的if条件中v != SelfNode为假，接着CAS操作失败，说明cxq队首已经变化，需要将v指向新队首
assert (_cxq != v, "invariant") ;
v = _cxq ; // CAS above failed - start scan at head of list
}
ObjectWaiter * p ;
ObjectWaiter * q = NULL ;
for (p = v ; p != NULL && p != SelfNode; p = p->_next) { // v指向cxq队首
q = p ;
assert (p->TState == ObjectWaiter::TS_CXQ, "invariant") ;
}
assert (v != SelfNode, "invariant") ;
assert (p == SelfNode, "Node not found on cxq") ;
assert (p != _cxq, "invariant") ;
assert (q != NULL, "invariant") ;
assert (q->_next == p, "invariant") ;
q->_next = p->_next ;
}
TEVENT (Unlink from cxq) ;
}
// Diagnostic hygiene ...
SelfNode->_prev = (ObjectWaiter *) 0xBAD ;
SelfNode->_next = (ObjectWaiter *) 0xBAD ;
SelfNode->TState = ObjectWaiter::TS_RUN ;
}
参考文献
http://www.diva-portal.org/smash/get/diva2:754541/FULLTEXT01.pdf
https://blog.csdn.net/penngrove/article/details/44175387
https://www.jianshu.com/p/1782e14a0766
https://pdfs.semanticscholar.org/edf9/54412a9b1ce955bea148199f325759779540.pdf
