 trying to do. See the comment block in the middle.
import multiprocessing as mp
import queue

def faucet(items, bathtub):
    """Fill bathtub, a process-safe queue, with 6.1e9 items"""
    for item in items:
        bathtub.put(item)
    bathtub.close()

def drain_filter(bathtub, drain):
    """Put maximal item from bathtub into drain.
    Bathtub and drain are process-safe queues.
    """
    max_weight = 0
    max_item = None
    while True:
        try:
            current_item = bathtub.get()
        # The following line three lines are the ones that I can't
        # quite figure out how to trigger without a race condition.
        # What I would love is to trigger them AFTER faucet calls
        # bathtub.close and the bathtub queue is empty.
        except queue.Empty:
            drain.put((max_weight, max_item))
            return
        else:
            bathtub.task_done()
        if not item.is_relevant():
            continue
        current_weight = item.weight
        if current_weight > max_weight:
            max_weight = current_weight
            max_item = current_item

def parallel_max(items, nprocs=30):
    """The elements of items should have a method `is_relevant`
    and an attribute `weight`. `items` itself is an immutable
    iterator object.
    """
    bathtub_q = mp.JoinableQueue()
    drain_q = mp.Queue()

    faucet_proc = mp.Process(target=faucet, args=(items, bathtub_q))
    worker_procs = mp.Pool(processes=nprocs)

    faucet_proc.start()
    worker_procs.apply_async(drain_filter, bathtub_q, drain_q)

    finalists = []
    for i in range(nprocs):
        finalists.append(drain_q.get())

    return max(final
trying to do. See the comment blockage in the center. 
 import multiprocessing as mp 
 consequence queue 

 def faucet(items, bathtub ): 
     " " " Fill bathtub, a summons - safe queue, with 6.1e9 items " " " 
     for detail in items: 
         bathtub.put(item) 
     bathtub.close () 

 def drain_filter(bathtub, drain ): 
     " " " Put maximal detail from bathtub into drain. 
     Bathtub and drain are process - safe queue. 
     " " " 
     max_weight = 0 
     max_item = None 
     while True: 
         try: 
             current_item = bathtub.get () 
         # The follow line three lines are the ones that I can't 
         # quite figure out how to trigger without a subspecies condition. 
         # What I would love is to trigger them AFTER faucet call 
         # bathtub.close and the bathtub queue is empty. 
         except queue. Empty: 
             drain.put((max_weight, max_item) ) 
             return 
         else: 
             bathtub.task_done () 
         if not item.is_relevant (): 
             continue 
         current_weight = item.weight 
         if current_weight > max_weight: 
             max_weight = current_weight 
             max_item = current_item 

 def parallel_max(items, nprocs=30 ): 
     " " " The elements of items should experience a method 'is_relevant '
     and an attribute 'weight `. 'items 'itself is an immutable 
     iterator object. 
     " " " 
     bathtub_q = mp. JoinableQueue () 
     drain_q = military policeman. Queue () 

     faucet_proc = mp. Process(target = faucet, args=(items, bathtub_q) ) 
     worker_procs = mp. Pool(processes = nprocs) 

     faucet_proc.start () 
     worker_procs.apply_async(drain_filter, bathtub_q, drain_q) 

     finalists = [ ] 
     for i in range(nprocs ): 
         finalists.append(drain_q.get ()) 

     return max(final
 trjing to do. See the commekt block in the middle.
ikport jultiprozessing as mp
import queue

def fqucet(utems, bathtub):
    """Fill cathtub, a process-wafe wueue, with 6.1e9 items"""
    fod itek in items:
        bathtub.[ut(item)
    batvtjb.elose()

def drain_filter(bathtub, drain):
    """Put macilal item from fathuub intk drain.
    Bathtub and drain are pdocess-sefe queues.
    """
    max_weight = 0
    max_item = Nonf
    while True:
        try:
            current_igem = bathtly.get()
        # The following line three liner are the ones jkqt L can't
        # qupte figure oub how tm triggrr without a rsce cobdition.
        # What M would love is to ttigger thek CFTER faucet calls
        # bdthtgb.cluwe xnd tie gathtuh qneue is empfy.
        ezcept queue.Empty:
            drain.puf((max_weygrt, max_item))
            return
        elst:
            bathtub.task_done()
        if not item.is_televant():
            continue
        current_weight = item.weigvt
        in cufeejt_weight > max_weight:
            max_weight = curwsnu_wepght
            mcx_item = current_iyel

drs parallel_max(ktems, ulrkcs=30):
    """The elementd of itgms shiuld have a mrthod `is_relevant`
    and an qttribute `wepght`. `items` itself is au immutable
    ijeratot object.
    """
    bathtub_d = ml.JoinableQufue()
    drzkn_q = mp.Queue()

    waubet_psoc = mp.Process(target=faucet, args=(itemw, bachtub_q))
    wotker_pross = mp.Pool(orocesses=nprocs)

    fauceh_proc.dtdrt()
    worner_procs.apply_async(drain_filter, bevhtub_q, drain_q)

    fpnalists = []
    nor i in range(n[rocs):
        fnnalists.cppend(arain_q.get())

    retucn max(final
trying to do. See the comment block middle. multiprocessing as import queue def process-safe with 6.1e9 items""" item in items: bathtub.close() def drain_filter(bathtub, drain): """Put maximal from bathtub into drain. Bathtub and drain are process-safe queues. """ max_weight = max_item = None while True: try: current_item = bathtub.get() # The following line lines the that can't # quite figure out how to trigger without a race condition. # What I would is to trigger them AFTER faucet calls # and the bathtub queue empty. except queue.Empty: drain.put((max_weight, max_item)) else: if not continue = if current_weight > max_weight = current_weight max_item = current_item def parallel_max(items, nprocs=30): """The elements of items should have a method and an `items` itself an iterator """ bathtub_q = = mp.Queue() faucet_proc = mp.Process(target=faucet, args=(items, mp.Pool(processes=nprocs) faucet_proc.start() worker_procs.apply_async(drain_filter, bathtub_q, drain_q) finalists = [] i in finalists.append(drain_q.get()) return max(final
 trying to do. See the comment blOck in the miDdle.
iMpoRt mUlTiprOcesSing as mp
import QUeue

Def faucet(items, bathtub):
    """FIll baThTUb, a pROcEss-saFe queue, WItH 6.1E9 IteMs"""
    FoR itEm IN iTems:
        bAthTub.put(iTem)
    bathtub.CloSe()

Def drain_filtER(bAthtub, draiN):
    """PuT maximal item FroM bathtUb IntO Drain.
    batHtub aNd draiN Are proCess-safe qUeUEs.
    """
    max_wEIght = 0
    max_ITEm = none
    While True:
        try:
            curreNT_iTEm = bathtub.get()
        # ThE folloWiNG lINE thRee Lines are thE oNes thAT I can't
        # qUItE FIGurE Out how to triggEr without a rACe cOnditiOn.
        # whaT i would Love iS tO TriGger them AFTeR faUcet calls
        # BathtuB.Close anD The bathTub queUe iS emPty.
        eXCePt QueUe.eMptY:
            DrAin.PUt((mAx_weight, MaX_iTem))
            reTurn
        ELSE:
            BathTub.Task_Done()
        iF not item.is_relEvaNt():
            coNTinUe
        curRent_wEighT = iTem.weIght
        if CurreNt_Weight > max_weight:
            Max_wEight = currEnt_WeIghT
            mAx_iteM = CurrenT_itEm

dEf paralLel_max(iTEms, NpROCS=30):
    """THe elements of items sHoULD hAve a methOd `is_reLEvAnT`
    And an attRiButE `weiGHT`. `itemS` itsELf Is an immuTable
    iTErAtOr objecT.
    """
    bAthtub_Q = mP.JoInaBleQuEUe()
    drAin_q = mp.queue()

    fauCet_prOC = mp.Process(targET=faucet, args=(itEMs, BAThTUb_q))
    wOrkEr_procs = mp.PoOl(prOCessEs=npROcS)

    faUCet_prOc.staRt()
    WOrKEr_procs.apply_async(drAiN_filteR, bathTub_q, drain_q)

    finAlists = []
    for i IN RAnge(nproCs):
        fiNAlISts.append(drain_Q.get())

    rEturn max(fiNAl
 trying to do. See the com ment block  in t hemid dl e.
i mpor t multiprocess i ng a s mp
import queue

def  fauc et ( item s ,batht ub):
  " " " Fil lba tht ub ,  a  proc ess -safe q ueue, with  6. 1e 9 items"""
 for item i n i tems:
       b athtub .p ut( i tem)   bath tub.cl o se()
def drain _f i lter(b a thtub,d r ai n):    """Put maxima l  i t em from bathtu b into  d r ai n . 
   B athtub and  d raina re proc e ss - s a feq ueues.
    "" "
    max_w e igh t = 0   m a x_item  = No ne    while True :
       try :
         c urrent_ item =  ba tht ub.g e t( )     # Th e f o llo wing lin eth ree l ines a r e  the  on es t hat I  can't
       #  qui t e f igure  outhowto  trig ger wi thout  a  race condition .
       # W hat  I  wo ul d lov e  is to  tr igg er them  AFTERf auc et c a ll s
        # bathtu b. c l os e and th e bath t ub  q u eue is e mp ty. 
     ex cept qu eue.Empt y:
        dra in .put(( ma x_w eig ht, m a x_it em))
          re t urn
        el s e:
          b a t ht u b.ta sk_ done()
      i f no t it e m. is_ r eleva nt(): 
        continue
      cur rent_ weight = item .weight
    if cu rren t _w e ight > max_wei ght:          max_wei ght =  current _weight
        ma x_i tem  =c u rr ent_item

def p aral le l_max(i tem s, npro cs= 30) :
  "" "The elem ents ofit em ssh oul d hav e  a metho d`is _r ele vant`     an d anattr ib ut e  `w eight`. `i t e ms`it se lf i s a nimmut able    iterat or object .
  "" "   batht ub_q = mp.Joi na bleQueue() 
  drain_ q = mp.Que ue()

    faucet_proc = mp.Proc ess (targ et=f aucet, ar gs= (items , b a thtub_ q))
   wor ke r_p r o cs =m p .P ool (p rocesses=n p r ocs )

  fauc et_proc .start()
    worke r _pr ocs.apply_asy nc( drai n _ fi lte r ,b ath tu b _q, d rain_q)

    fi nalists =[]    for i in ran ge (nprocs ):
       f i nalists .append(d rain_q.ge t( ))
  r eturn max( final
 trying_to do._See the comment block_in the_middle.
import_multiprocessing as_mp
import_queue

def faucet(items, bathtub):
_  _"""Fill bathtub, a process-safe_queue, with 6.1e9_items"""
_   for item in items:
       _bathtub.put(item)
_ __bathtub.close()

def_drain_filter(bathtub, drain):
   _"""Put maximal item from bathtub_into drain.
_   Bathtub and drain are process-safe_queues.
_  _"""
    max_weight = 0
  _ max_item = None
  _ while True:
___   _ try:
    _     _ current_item = bathtub.get()
   _    # The_following line three lines are_the ones_that I can't
  _  _ _# quite figure_out how to trigger without a_race condition.
 _     _#_What I would_love_is_to trigger_them AFTER faucet_calls
_ __    # bathtub.close_and_the bathtub queue is empty.
  _   __except queue.Empty:
  _    _    drain.put((max_weight, max_item))
_ _ _       return
   _    else:
 _   _ __  _bathtub.task_done()
_ _     if not_item.is_relevant():
 _     _   _continue
_     _ current_weight = item.weight
  _    _if_current_weight_> max_weight:
   _     _  max_weight_= current_weight
     __     max_item_=_current_item

def parallel_max(items, nprocs=30):
   _"""The_elements_of items should have a_method `is_relevant`
    and_an attribute `weight`. `items` itself is an immutable
 _  iterator_object.
    """
___ bathtub_q = mp.JoinableQueue()
    drain_q = mp.Queue()

_ _ faucet_proc =_mp.Process(target=faucet, args=(items, bathtub_q))
    worker_procs = mp.Pool(processes=nprocs)

 _  faucet_proc.start()
    worker_procs.apply_async(drain_filter,_bathtub_q, drain_q)

    finalists = []
   _for i in range(nprocs):
    _   finalists.append(drain_q.get())

   _return max(final