ó ßüÚ\c@`sģddlmZmZmZddlmZddlZddlZddlm Z m Z ddl m Z ddl mZddlmZejeƒZd efd „ƒYZdS( i(tprint_functiontdivisiontabsolute_import(tdequeN(tgentlocks(tIOLoopi(tCommClosedError(tparse_timedeltat BatchedSendcB`steZdZd d d„Zd„Zd„Zd„ZeZe j d„ƒZ d„Z e j d„ƒZ d„ZRS( sq Batch messages in batches on a stream This takes an IOStream and an interval (in ms) and ensures that we send no more than one message every interval milliseconds. We send lists of messages. Batching several messages at once helps performance when sending a myriad of tiny messages. Example ------- >>> stream = yield connect(ip, port) >>> bstream = BatchedSend(interval='10 ms') >>> bstream.start(stream) >>> bstream.send('Hello,') >>> bstream.send('world!') On the other side, the recipient will get a message like the following:: ['Hello,', 'world!'] cC`sē|ptjƒ|_t|ddƒ|_tjƒ|_tjƒ|_t |_ g|_ d|_ d|_d|_d|_d|_tdtjjdƒƒ|_||_dS(Ntdefaulttmsitmaxlens+distributed.comm.recent-messages-log-length(RtcurrenttloopRtintervalRtEventtwakertstoppedtFalset please_stoptbuffertNonetcommt message_countt batch_countt byte_countt next_deadlineRtdasktconfigtgettrecent_message_logt serializers(tselfRRR ((s2lib/python2.7/site-packages/distributed/batched.pyt__init__*s       cC`s ||_|jj|jƒdS(N(RRt add_callbackt_background_send(R!R((s2lib/python2.7/site-packages/distributed/batched.pytstart<s cC`s|jo|jjƒS(N(Rtclosed(R!((s2lib/python2.7/site-packages/distributed/batched.pyR&@scC`s%|jƒrdSdt|jƒSdS(Nss(R&tlenR(R!((s2lib/python2.7/site-packages/distributed/batched.pyt__repr__Cs cc`s”x€|js‚y%|jj|jƒV|jjƒWntjk rGnX|js`d|_qn|jdk r|j j ƒ|jkrqn|jg}|_|j d7_ |j j ƒ|j |_z­yd|j j|d|jddƒV}|dkr|jj|ƒn|jjdƒ|j|7_WnBtk rU}tjd|ƒPntk rstjdƒPnXWdd}XqW|jjƒdS( NiR ton_errortraiseg€„.As large-messagesBatched Comm Closed: %ssError in batched write(RRtwaitRtclearRt TimeoutErrorRRRttimeRRRtwriteR RtappendRRtloggertinfot Exceptiont exceptionRtset(R!tpayloadtnbyteste((s2lib/python2.7/site-packages/distributed/batched.pyR$Ks<   '      cC`si|jdk r'|jjƒr't‚n|jd7_|jj|ƒ|jdkre|jj ƒndS(sl Schedule a message for sending to the other side This completes quickly and synchronously iN( RRR&RRRR0RRR5(R!tmsg((s2lib/python2.7/site-packages/distributed/batched.pytsendqs  cc`sē|jdkrdSt|_|jjƒ|jjƒV|jjƒsŪyC|j rˆg|j |_ }|jj |d|j ddƒVnWnt k rœnX|jj ƒVndS(s- Flush existing messages and then close comm NR R)R*(RRtTrueRRR5RR+R&RR/R Rtclose(R!R6((s2lib/python2.7/site-packages/distributed/batched.pyR<s     cC`sU|jdkrdSt|_g|_|jjƒ|jjƒsQ|jjƒndS(N( RRR;RRRR5R&tabort(R!((s2lib/python2.7/site-packages/distributed/batched.pyR=’s   N(t__name__t __module__t__doc__RR"R%R&R(t__str__Rt coroutineR$R:R<R=(((s2lib/python2.7/site-packages/distributed/batched.pyR s   & (t __future__RRRt collectionsRtloggingRttornadoRRttornado.ioloopRtcoreRtutilsRt getLoggerR>R1tobjectR (((s2lib/python2.7/site-packages/distributed/batched.pyts