U ҥc|P@sdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddlm Z m Z ddlmZmZmZmZddlmZddlZddlmZdd lmZdd lmZdZd ZGd d d ZGdddZGddde Z dS)z.Wrappers for forwarding stdout/stderr over zmqN)b2a_hex)deque)StringIO TextIOBase)AnyCallableDequeOptional)WeakSet)extract_header)IOLoop) ZMQStreamc@seZdZdZd&ddZddZddZed d Zd d Z d dZ ddZ ddZ ddZ ddZddZddZddZeddZdd Zd!d"Zd#d$Zd%S)' IOPubThreadaAn object for sending IOPub messages in a background thread Prevents a blocking main thread from delaying output from threads. IOPubThread(pub_socket).background_socket is a Socket-API-providing object whose IO is always run in a thread. FcCs||_t||_t|_||_tdd|_|r8| t |_ t |_t|_|t j|jdd|_d|j_d|j_d|j_d|j_dS)aCreate IOPub thread Parameters ---------- socket : zmq.PUB Socket the socket on which messages will be sent. pipe : bool Whether this process should listen for IOPub messages piped from subprocesses. F)Z make_currentIOPub)targetnameTN)socketBackgroundSocketZbackground_socketosgetpid _master_pid _pipe_flagr io_loop_setup_pipe_in threadinglocal_localr_eventsr _event_pipes_setup_event_pipeThread _thread_mainthreaddaemonZpydev_do_not_traceZis_pydev_daemon_threadr)selfrpiper'1lib/python3.8/site-packages/ipykernel/iostream.py__init__-s     zIOPubThread.__init__cCs|j|jjdddS)z.The inner loop that's actually run in a threadT)Zall_fdsN)rstartcloser%r'r'r(r"Is zIOPubThread._thread_maincCsf|jj}|tj}d|_ttdd}d|}|_ | |t ||j |_ |j |jdS)zLCreate the PULL socket listening for events that should fire in this thread.rasciiz inproc://%sN)rcontextzmqPULLlingerrrurandomdecode_event_interfaceZbindr rZ _event_pulleron_recv _handle_event)r%ctxpipe_inZ_uuidZifacer'r'r(r Ns  zIOPubThread._setup_event_pipecCs`z |jj}WnNtk rZ|jj}|tj}d|_||j ||j_|j |YnX|S)zSthread-local event pipe for signaling events that should be processed in the threadr) r event_pipeAttributeErrorrr/r0PUSHr2connectr5radd)r%r:r8r'r'r( _event_pipeZs   zIOPubThread._event_pipecCs,t|j}t|D]}|j}|qdS)zHandle an event on the event pipe Content of the message is ignored. Whenever *an* event arrives on the event stream, *all* waiting events are processed in order. N)lenrrangepopleft)r%msgZn_events_Zevent_fr'r'r(r7ks   zIOPubThread._handle_eventc Cs|jj}td|_|tj}d|_z|d|_ WnJtj k r}z*t d|dd|_ |WYdSd}~XYnXt||j|_|j|jdS)z7setup listening pipe for IOPub from forked subprocessesr-rztcp://127.0.0.1z)Couldn't bind IOPub Pipe to 127.0.0.1: %sz' subprocess output will be unavailable.FN)rr/rr3 _pipe_uuidr0r1r2Zbind_to_random_port _pipe_portZZMQErrorwarningswarnrr+r rZ_pipe_inr6_handle_pipe_msg)r%r8r9er'r'r(rzs"  zIOPubThread._setup_pipe_incCsJ|jr|sdS|d|jkr4td|tjddS||dddS)z'handle a pipe message from a subprocessNrzBad pipe message: %sfiler)r_is_master_processrEprintsys __stderr__send_multipartr%rCr'r'r(rIs zIOPubThread._handle_pipe_msgcCs2t}|tj}d|_|d|j||fS)Ni ztcp://127.0.0.1:%i)r0ZContextrr<r2r=rF)r%r8pipe_outr'r'r(_setup_pipe_outs  zIOPubThread._setup_pipe_outcCst|jkSNrrrr,r'r'r(rMszIOPubThread._is_master_processcCs|jr|rtStSdS)z8check for forks, and switch to zmq pipeline if necessaryN)rrMMASTERCHILDr,r'r'r(_check_mp_modeszIOPubThread._check_mp_modecCs"d|j_|jt|jdS)zStart the IOPub threadrN)r#rr*atexitregisterstopr,r'r'r(r*s zIOPubThread.startcCs@|jsdS|j|jj|j|jD] }|q.dS)zStop the IOPub threadN)r#is_aliverZ add_callbackr\joinrr+)r%r:r'r'r(r\s    zIOPubThread.stopcCs|jr dS|jd|_dS)zClose the IOPub thread.N)closedrr+r,r'r'r(r+s zIOPubThread.closecCs |jdkSrU)rr,r'r'r(r_szIOPubThread.closedcCs.|jr$|j||jdn|dS)ztSchedule a function to be called in our IO thread. If the thread is not running, call immediately. N)r#r]rappendr?send)r%fr'r'r(schedules  zIOPubThread.schedulecsfdddS)zsend_multipart schedules actual zmq send in my thread. If my thread isn't running (e.g. forked process), send immediately. cs jSrU) _really_sendr'argskwargsr%r'r(r`z,IOPubThread.send_multipart..N)rdr%rgrhr'rfr(rQszIOPubThread.send_multipartcOsj|jr dS|}|tkr0|jj|f||n6|\}}|j|jf|f||||dS)z)The callback that actually sends messagesN) r_rYrXrrQrTrEr+Zterm)r%rCrgrhZmp_moder8rSr'r'r(res zIOPubThread._really_sendN)F)__name__ __module__ __qualname____doc__r)r"r propertyr?r7rrIrTrMrYr*r\r+r_rdrQrer'r'r'r(r$s(       rcsHeZdZdZdZddZfddZfddZd d Zd d Z Z S) rz>Wrapper around IOPub thread that provides zmq send[_multipart]NcCs ||_dS)zInitialize the socket.N) io_thread)r%rpr'r'r(r)szBackgroundSocket.__init__csv|dr |dr t||jdk s.tt|jj|rftj d|d|t ddt |jj|St|dS)z2Wrap socket attr access for backward-compatibility__NzAccessing zmq Socket attribute O on BackgroundSocket is deprecated since ipykernel 4.3.0 use .io_thread.socket. stacklevel) startswithendswithsuper __getattr__rpAssertionErrorhasattrrrGrHDeprecationWarninggetattr)r%attr __class__r'r(rys zBackgroundSocket.__getattr__csj|dks|dr,|dr,t||n:tjd|d|tdd|jdk sVtt |jj ||dS)zSet an attribute on the socket.rprqzSetting zmq Socket attribute rrrsrtN) rvrwrx __setattr__rGrHr|rprzsetattrr)r%r~valuerr'r(r szBackgroundSocket.__setattr__cOs|j|gf||S)zSend a message to the socket.)rQ)r%rCrgrhr'r'r(rbszBackgroundSocket.sendcOs|jdk st|jj||S)zSchedule send in IO threadN)rprzrQrjr'r'r(rQszBackgroundSocket.send_multipart) rkrlrmrnrpr)ryrrbrQ __classcell__r'r'rr(rs  rc@seZdZUdZdZdZdZdZdZe e e d<ddZ d d Z d-d d d ddZddZddZddZddZddZeddZddZddZd d!Zee ed"d#d$Zd%d&Zd'd(Zd)d*Zd+d,ZdS). OutStreamzpA file like object that publishes the stream to a 0MQ PUB socket. Output is handed off to an IO Thread g?NzUTF-8_exccCs(t|dddk r|jSd}t|dS)z^ Things like subprocess will peak and write to the fileno() of stderr/stdout. _original_stdstream_copyNfileno)r}rioUnsupportedOperationrRr'r'r(r4szOutStream.filenocCslzHt|jd}|rF|jrF||t|j|t|jd}qWntk rft |_ YnXdS)ai We've redirected standards steams 0 and 1 into a pipe. We need to watch in a thread and redirect them to the right places. 1) the ZMQ channels to show in notebook interfaces, 2) the original stdout/err, to capture errors in terminals. We cannot schedule this on the ioloop thread, as this might be blocking. iN) rread_fid _should_watchwriter4r ExceptionrOexc_infor)r%Zbtsr'r'r(_watch_pipe_fd>s  zOutStream._watch_pipe_fdTF)watchfdisattyc Cs|dk rtjdtdd||_t|tsLtjd|tddt|}|||_||_d| |_ i|_ t |_d|_d|_|j|_t|_t|_d|_t||_d|_|rtjdstjd rd t jkrd |_| ||rt!|d rt!|d r||_n d}t"|dS)a Parameters ---------- session : object the session object pub_thread : threading.Thread the publication thread name : str {'stderr', 'stdout'} the name of the standard stream to replace pipe : object the pip object echo : bool whether to echo output watchfd : bool (default, True) Watch the file descripttor corresponding to the replaced stream. This is useful if you know some underlying code will write directly the file descriptor by its number. It will spawn a watching thread, that will swap the give file descriptor for a pipe, read from the pipe, and insert this into the current Stream. isatty : bool (default, False) Indication of whether this stream has termimal capabilities (e.g. can handle colors) NzKpipe argument to OutStream is deprecated and ignored since ipykernel 4.2.3.rsrtzISince IPykernel 4.3, OutStream should be created with IOPubThread, not %rsstream.FZlinuxdarwinZPYTEST_CURRENT_TESTTrrz(echo argument must be a file like object)#rGrHr|session isinstancerr* pub_threadrencodetopic parent_headerrrr_flush_pending_subprocess_flush_pendingr_io_looprRLock _buffer_lockr_bufferechobool_isattyrrOplatformrvenviron_setup_stream_redirectsr{ ValueError) r%rrrr&rrrrCr'r'r(r)TsZ"       zOutStream.__init__cCs|jS)zpReturn a bool indicating whether this is an 'interactive' stream. Returns: Boolean )rr,r'r'r(rszOutStream.isattycCsdt\}}tt|}t||_t||||_d|_ t j |j d|_ d|j _|j dS)N)rT)rr&r}rOrduprdup2rrrr!rwatch_fd_threadr$r*)r%rZprZpwZfnor'r'r(rs   z!OutStream._setup_stream_redirectscCst|jkSrUrVr,r'r'r(rMszOutStream._is_master_processcCst||_dS)zSet the parent header.N)r r)r%parentr'r'r( set_parentszOutStream.set_parentcCs@|jrd|_|j|jr6|j\}}}t|||d|_dS)zClose the stream.FN)rrr^r tracebackprint_exceptionr)r%etypertbr'r'r(r+s  zOutStream.closecCs |jdkSrU)rr,r'r'r(r_szOutStream.closedcs,jr dSd_fdd}j|dS)zuschedule a flush in the IO thread call this on write, to indicate that flush should be called soon. NTcsjjjdSrU)rZ call_laterflush_interval_flushr'r,r'r(_schedule_in_threadsz6OutStream._schedule_flush.._schedule_in_thread)rrrd)r%rr'r,r(_schedule_flushs  zOutStream._schedule_flushcCs~|jrr|jjdk rr|jjrr|jjjtjkrr|j|jt}|j|j | |j szt dt jdn|dS)zStrigger actual zmq send send will happen in the background thread NzIOStream.flush timed outrK)rr#r]identrcurrent_threadrdrEventsetwait flush_timeoutrNrOrP)r%Zevtr'r'r(flushs   zOutStream.flushc Csd|_d|_|jdk rhz|jWnBtk rf}z$|jtjk rVtd|tjdW5d}~XYnX|}|rt |j _ |j |d}|j j|jd||j|jddS)zThis is where the actual send happens. _flush should generally be called in the IO thread, unless the thread has been destroyed (e.g. forked subprocess). FNzFlush failed: rK)rtextstream)contentrr)rrrrOSErrorrOrPrN _flush_bufferrrrpidrrbrrr)r%rJdatarr'r'r(rs&  &  zOutStream._flush)stringreturnc Cst|ts dt|}t||jdk r~z|j|WnBtk r|}z$|jtjk rlt d|tjdW5d}~XYnX|j dkrd}t |nT| }|j |j|W5QRX|r|jrdSd|_|j |jn|t|S)zWrite to current stream after encoding if necessary Returns ------- len : int number of items from input parameter written to stream. z"write() argument must be str, not NzWrite failed: rKI/O operation on closed fileT)rstrtype TypeErrorrrrrOrPrNrrrMrrrrdrrr@)r%rrCrJZis_childr'r'r(rs,   &   zOutStream.writecCs0|jdkrd}t|n|D]}||qdS)zWrite lines to the stream.Nr)rrr)r%ZsequencerCrr'r'r( writelinesAs   zOutStream.writelinescCsdS)z$Test whether the stream is writable.Tr'r,r'r'r(writableJszOutStream.writablecCs|}|}||S)zs,      O3