architecture of so-ware systems · lecture 5: threads, synchronizaon, atomic operaons, non-blocking...

Architectureofso-waresystems

Lecture5:Threads,synchroniza:on,atomicopera:ons,non-blockingalgorithms

David Šišlák [email protected]

Threads

»  processesvs.threads»  bothsupportconcurrentexecu:on»  oneprocesshasoneormul:plethreads»  threadssharethesameaddressspace(dataandcode)»  contextswitchingbetweenthreadsisusuallylessexpensive»  threadinter-communica:onisrela:velyefficientusingsharedmemory

»  athreadexecutessequenceofcodewithownstackwithframes t.getStackTrace()

»  ownlocalvariables»  ownmethodparameters

»  threadcrea:onby»  subclassofjava.lang.Thread»  implementa:onofjava.lang.Runnable

24/05/17 2A4B77ASS–Lecture5

Threads

»  Thread.currentThread()

»  eachthreadhas»  id–uniquelongid,onlyget»  name-get/set»  priority–get/set

»  Thread.MIN_PRIORITY(1),NORM_PRIORITY(5),MAX_PRIORITY(10)»  threadgroup–get/set»  uncaughtexcep:onhandler–get/set+get/setDefaultExcep:onHandler

»  UncaughtExcep:onHandler»  daemonflag–is/set»  contextclassloader–get/set,usedtoloadclassesandresourcesinside»  interrupted–interrupt(),isInterrupted(),sta:cinterrupted()

»  InterruptedExcep:on»  status–seenextslide


Threads

»  threadstates-t.getState()»  new

»  a-ercrea:on»  runnable

»  start()»  blocked

» wai:ngforalock,(re)entersynchronizedmethod/block»  wai9ng(canbeinterrupted)

»  o.wait(),t.join(),LockSupport.park()»  9medwai9ng(canbeinterrupted)

»  Thread.sleep(x),o.wait(x),t.join(x)»  LockSupport.parkNanos(x),LockSupport.parkUn:l(:me)

»  terminated»  finishedt.run()method,Run:me.exit(),t.stop()

»  Thread.yield()–allowotherthreadstoexecute(nonpreemp:veOS)24/05/17 4A4B77ASS–Lecture5

Threads–usageofRunnable


Threads–subclassofThread


ExecutorService

»  conceptofthreadpoolingsince1.5»  suitableforexecu:onoflargenumberofasynchronoustasks

•  e.g.HTTPrequestsinserver»  reduceoverheadwithThreadcrea:onforeachtask,contextswitching»  interface-java.u:l.concurrent.ExecutorService

•  shutdown(),shutdownNow(),awaitTermina:on•  execute(Runnabler)•  Future<?>submit(Runnabler),Future<T>submit(Callable<T>c)

»  java.u:l.concurrent.Future<T>•  booleancancel(booleanmayInterruptIfRunning)•  isCancelled(),isDone()•  Vget(),Vget(long:meout,TimeUnitunit)

»  java.u:l.concurrent.Executors(op:onallywithThreadFactory)•  newSingleThreadExecutor()•  newFixedThreadPool(nThreads)•  newCachedThreadPool()–default60secondskeep-alive


Dataraces

»  r1andr2arelocalvariables»  AandBaresharedvariables(heaplocated)ini:allysetto0

»  whatcanbetheresultsforr1andr2?

24/05/17 ?8A4B77ASS–Lecture5

Thread1 Thread2

r2=A; r1=B;

B=1; A=2;

Dataraces

»  r1andr2arelocalvariables»  AandBaresharedvariables(heaplocated)ini:allysetto0

»  whatcanbetheresultsforr1andr2?•  r1=0,r2=0•  r1=1,r2=0•  r1=0,r2=2•  anythingelse?


Thread1 Thread2

r2=A; r1=B;

B=1; A=2;

Synchronized

»  eachobjectisassociatedwithmonitor»  synchronizedisimplementedusingmonitors

»  isthiscorrect?


Synchronized

»  eachobjectisassociatedwithmonitor»  synchronizedisimplementedusingmonitors


Reentrantlocks

»  java.u:l.concurrent.locks.ReentrantLocksince1.5»  extendedopera:onsincomparisontosynchronized:

•  lock(),unlock()•  lockInterrup:bly()throwsInterruptedExcep:on•  booleantryLock()•  booleantryLock(long:meout,TimeUnitunit)throws

InterruptedExcep:on»  fairness

•  newReentrantLock(booleanfair),bydefaultunfair•  synchronizedisunfair!•  fairlocksareslower!


SynchronizedVS.reentrantlock

»  isthiscorrecttransforma:on?


SynchronizedVS.reentrantlock

»  isthiscorrecttransforma:on?»  NO–needcatchexcep:ons


Reentrantlockperformance–testexample

24/05/17 15A4B77ASS–Course6

Reentrantlockperformance–testexamplecont.


Reentrantlockperformance-results

»  Javaon2xIntelXeonE54202.5GHz(8coresintotal)

»  resultsfor8threads:

LOCKopera:ons/second3499925SYNCopera:ons/second1104862LOCKopera:ons/second3478742SYNCopera:ons/second1149406LOCKopera:ons/second3500417SYNCopera:ons/second1121584

»  butReentrantLockisstandardobjectonheap


Deadlock-example


Threadruncontrol–synchronizedVSreentrantlock


»  isthiscorrect?

Threadruncontrol–stoppabletask


»  isthiscorrect?»  NO–missingvola:le

Threadruncontrol–stoppabletask


Vola9levariable

»  nevercachedthread-locally–allaccessdirectlytomainmemory»  guaranteesatomicreadandwriteopera:ons(definesmemorybarrier)»  canbeusedforbothprimi:vesandobjects(references)»  don’tblockthreadexecu:on


Vola9levariable

»  nevercachedthread-locally–allaccessdirectlytomainmemory»  guaranteesatomicreadandwriteopera:ons(definesmemorybarrier)»  canbeusedforbothprimi:vesandobjects(references)»  don’tblockthreadexecu:on»  usefulforone-threadwrite»  notsuitableforread-update-writeopera9ons

»  notnecessaryfor:•  immutableobjects•  variableaccessedbyonlyonethread•  wherevariableiswithincomplexsynchronizedopera:on


»  whatvalueisreturnedwhilecallingreader()?

Vola9leexample–mul9-CPU



»  memorychangesmadebyoneCPUcanbepropagatedbacktomainmemoryout-of-order

»  correctbehaviorsinceJAVA1.5wherememorybarrierisusedforvola:le•  usageofspecificCPUinstruc:ontoguaranteeit



»  memorychangesmadebyoneCPUcanbepropagatedbacktomainmemoryout-of-order

»  correctbehaviorsinceJAVA1.5wherememorybarrierisusedforvola:le•  usageofspecificCPUinstruc:ontoguaranteeit

»  butwhatformul:-threadwriteandread-update-writeopera:ons?•  synchroniza:on/reentrantlocks•  atomicopera:ons


Atomicopera9ons

»  specificCPUinstruc:on–CMPXCHGcompare-and-exchangeorCAS(compare-and-swap)

»  32-bitCPUssupport64-bitCASopera:ons»  64-bitCPUssupport128-bitCASopera:ons»  JAVAu:lizesonly64-bitversioninjava.u:l.concurrent.atomic:

•  AtomicBoolean•  AtomicInteger•  AtomicLong•  AtomicReference

»  basicopera:onsinAtomicInteger:•  intget(),set(intvalue),booleancompareAndSet(intexpect,intupdate)•  intaddAndGet(intdelta)•  intincrementAndGet(),intdecrementAndGet()


Atomiccounter

»  howistheatomicincrementAndGetimplementedusingCASinstruc:on?


AtomicInteger–implementa9on


»  isaccesstoarraymemberasvola:le?

Vola9learrays


Vola9learrays–solu9on1

»  atomicarrayversions:•  AtomicIntegerArray•  AtomicLongArray•  AtomicReferenceArray

»  basicopera:onsforAtomicIntegerArray:•  intget(inti),set(inti,intnewValue)-providesvola/leaccesstomember•  booleancompareAndSet(inti,intexpectedValue,intnewValue)•  intincrementAndGet(inti),intdecrementAndGet(inti)


Vola9learrays–solu9on2(>=Java1.5)

»  donotrequirewrapperobject»  butslightlyinefficientduetoanotherread-writeopera:on»  donotsupportread-update-writeopera:ons


Atomicfieldupdaters

»  suitablewithlargenumberofobjectofthegiventype–itsavesmemory•  don’trequiresingleinstancetohaveanextraobjectembedded

»  refervariable“normally”withoutgeuerandseuers


Atomicfieldupdaters

»  butbewareoflessefficientopera:onsoveratomicfieldupdaters»  AtomicIntegerFieldUpdater:

»  exis:ngfieldupdaters:•  AtomicIntegerFieldUpdater•  AtomicLongFieldUpdater•  AtomicReferenceFieldUpdater

»  noarrayfieldupdaterexists


Atomiccomplextypes

»  AtomicMarkableReference•  objectreferencealongwithamarkbit

»  AtomicStampedReference•  objectreferencealongwithaninteger“stamp”

24/05/17 36A4B77ASS–Course4

Atomiccomplextypes

»  AtomicMarkableReference•  objectreferencealongwithamarkbit

»  AtomicStampedReference•  objectreferencealongwithaninteger“stamp”

»  notes:•  usefulforABAproblem

‒  A->BandB->A,howcanIknowthatAhasbeenchangedsincethelastobserva:on?

•  doesn’tusedouble-wideCAS(CAS2,CASX)->muchslowerthansimpleatomictypesduetoobjectalloca9on


Non-blockingalgorithms

»  lock-free,some:meswait-free,basedonCASinstruc:ons

»  sharedresourcessecuredbylocks:•  high-prioritythreadcanbeblocked(e.g.interrupthandler)•  parallelismreducedbycoarse-grainedlocking(unfairlocks)•  fine-grainedlockingandfairlocksincreasesoverhead•  canleadtodeadlocks,priorityinversion(low-prioritythreadholdsa

sharedresourcewhichisrequiredbyhigh-prioritythread)

»  non-blockingalgorithmsproper9es:•  outperformblockingalgorithmsbecausemostofCASsucceedsonthe

firsttry•  removescostforsynchroniza:on,threadsuspension,contextswitching

»  note:requiredforreal-:mesystems


Non-blockingstack(LIFO)

»  basedonTreiber’salgorithm(1986)


Thread-safecollec9onsandmaps

»  blockingvariants:•  sta:c<T>Collec:on<T>Collec:ons.synchronizedCollec:on(Collec:on<T>c)•  sta:c<T>List<T>Collec:ons.synchronizedList(List<T>list)•  sta:c<K,V>Map<K,V>Collec:ons.synchronizedMap(Map<K,V>m)•  sta:c<T>Set<T>Collec:ons.synchronizedSet(Set<T>s)•  alsoforSortedSetandSortedMap

»  non-blockingvariants:•  ConcurrentLinkedQueue(interfaceCollec:on,Queue):

‒  Epeek(),Epoll(),add(E)•  ConcurrentHashMap(interfaceMap):

‒  putIfAbsent(Kkey,Vvalue),remove(Objectkey,Objectvalue)‒  replace(Kkey,VoldValue,VnewValue)

•  ConcurrentSkipListMap(interfaceSortedMap),ConcurrentSkipListSet(interfaceSortedSet)


architecture of so-ware systems · lecture 5: threads, synchronizaon, atomic operaons, non-blocking...

Documents