&GRIf@sddlZddlZddlZddlZddlZddlZddlmZmZm Z ddlm Z m Z m Z ddlm Z mZddlmZmZmZmZddlmZmZmZmZmZddlmZmZddlmZmZmZdd lmZm Z d Z!d Z"d Z#d Z$dZ%dZ&dZ'dZ(d Z)GdddeZ*Gdddej+Z,GdddZ-Gdddej+Z.Gddde.Z/Gddde.Z0dS)!N) basestringbytesstr)Property getErrorMsgUID) PackageQueueTinyPackageQueue)PackagePACKAGE_HEARTBEAT PACKAGE_CLOSEEINTR)can_sendcan_recvsend_allrecv_all HEADER_SIZE) Connection TIMEOUT_MIN) STATUS_CLOSEDSTATUS_WAITINGSTATUS_HOSTING)STATUS_CONNECTEDSTATUS_CLOSINGz Socket error.z!Other end dropped the connection.zHandshake timed out.zHandshake failed.z2Handshake failed (context cannot connect to self).zClosed from other end.zLost track of the stream.zError in io thread. ic@seZdZdZdddZdddZdd d Zd d d ZddZdddZ ddZ ddZ dS) TcpConnectionar TcpConnection(context, name='') The TcpConnection class implements a connection between two contexts that are in differenr processes or on different machines connected via the internet. This class handles the low-level communication for the context. A ContextConnection instance wraps a single BSD socket for its communication, and uses TCP/IP as the underlying communication protocol. A persisten connection is used (the BSD sockets stay connected). This allows to better distinguish between connection problems and timeouts caused by the other side being busy. cCs>d|_d|_td|j|_tj|||dS)N@)_sendingThread_receivingThreadr Z _queue_params_qoutr__init__)selfcontextnamer%O/Applications/pyzo2014a/lib/python3.4/site-packages/iep/yoton/connection_tcp.pyr!:s  zTcpConnection.__init__NcCs]|jj|dk r[|j\|_|_|tkr[|j\|_|_q[nt j ||z|t t gkr||_ |jdt||_t||_|jj|jjn|dkrGy|j jWntk rYnXy|j jWntk r(YnXd|_ d|_d|_nWd|jjXdS)a  _connected(status, bsd_socket=None) This method is called when a connection is made. Private method to apply the bsd_socket. Sets the socket and updates the status. Also instantiates the IO threads. Nr)_lockacquire getsockname _hostname1_port1r getpeername _hostname2_port2r _set_statusrr _bsd_socket setblocking SendingThreadrReceivingThreadrstartshutdown Exceptioncloserelease)r"status bsd_socketr%r%r&r0Gs6           zTcpConnection._set_statusr'c Cs>tjtjtj}|jtjtjt|jtjtjttj j dsx|jtjtj dnxt |||D]H}y|j ||fPWqtk r|dkrnwYqXqWt|}d|d}t||jd|jt|t|||_|jjdS)z Bind the bsd socket. Launches a dedicated thread that waits for incoming connections and to do the handshaking procedure. winr'zCould not bind to any of the z ports tried.rN)socketAF_INET SOCK_STREAM setsockopt SOL_SOCKET SO_SNDBUFSOCKET_BUFFERS_SIZE SO_RCVBUFsysplatform startswith SO_REUSEADDRrangebindr7rIOErrorlistenr0r HostThreadZ _hostThreadr5)r"hostnameport max_triessport2tmpr%r%r&_binds(      zTcpConnection._bindg?c Cstjtjtj}|jtjtjt|jtjtjt|dkr_d}nd}tj|}xw| rtj|kry|j ||fd}Wn)tj k rYntj k rYnXtj |dqxW|s5t j\}}} ~ t|} td||| fnt|} | j|j\} } | s|jd| s{d} ntd| n| \|_|_|jt|d S) z$ Connect to a bound socket. g{Gz?FTgY@zCannot connect to %s on %i: %srzproblem during handshakezCould not connect: N)r=r>r?r@rArBrCrDtimeconnecterrortimeoutsleeprEexc_inforrK HandShakershake_hands_as_clientid1r0_id2_pid2r)r"rNrOrXrQok timestamptypevaluetberrhsuccessinfor%r%r&_connects<       zTcpConnection._connectc Cs:|jjty|jdWntk r5YnXdS)z Send PACKAGE_CLOSE. g?N)r pushr flushr7)r"r%r%r&_notify_other_end_of_closings  z*TcpConnection._notify_other_end_of_closingg@cCs|jstdn|jjttj|}xK|jr|jj rtjdtj|kr9tdq9q9WdS)zn Put a dummy message on the queue and spinlock until the thread has popped it from the queue. zCannot flush if not connected.g{Gz?zSending the packages timed out.N) is_connected RuntimeErrorr rjr rUemptyrY)r"rXrar%r%r&_flushs   zTcpConnection._flushcCs|jj|dS)zU Put package on the queue, where the sending thread will pick it up. N)r rj)r"packager%r%r& _send_packageszTcpConnection._send_packagecCs|jjj|dS)z> Put package in queue, but bypass potential blocking. N)r _qappend)r"rqr%r%r&_inject_packageszTcpConnection._inject_package) __name__ __module__ __qualname____doc__r!r0rTrirlrprrrur%r%r%r&r*s  D95  rc@s:eZdZdZddZddZddZdS) rMa HostThread(context_connection, bds_socket) The host thread is used by the ContextConnection when hosting a connection. This thread waits for another context to connect to it, and then performs the handshaking procedure. When a successful connection is made, the context_connection's _connected() method is called and this thread then exits. cCs3tjj|||_||_|jddS)NT) threadingThreadr!_context_connection_bsd_host_socket setDaemon)r"context_connectionr;r%r%r&r!2s  zHostThread.__init__cCsx|jjr|j}|s'qn|jjs7Pnt|}|j|jj\}}|r|d|j_|d|j_ntd|q|j j |jj t |PqW|`|` dS)z run() The main loop. Waits for a connection and performs handshaking if successfull. rr'zYoton: Handshake failed: N) r| is_waiting_wait_for_connectionr[shake_hands_as_hostr]r^r_printr}r8r0r)r"rQZhsrgrhr%r%r&run=s$     zHostThread.runc Cs|jjdx|jjry|jj\}}|SWqtjk rSYqtjk rtj \}}}~|j t krnYqXqWdS)z _wait_for_connection() The thread will wait here until someone connects. When a connections is made, the new socket is returned. g?N) r} settimeoutr|racceptr=rXrWrErZerrnor )r"rQaddrrbrcrdr%r%r&rhs zHostThread._wait_for_connectionN)rvrwrxryr!rrr%r%r%r&rM&s  +rMc@sUeZdZdZddZddZddZdd d Zd d Zd S)r[a HandShaker(bsd_socket) Class that performs the handshaking procedure for Tcp connections. Essentially, the connecting side starts by sending 'YOTON!' followed by its id as a hex string. The hosting side responds with the same message (but with a different id). This process is very similar to a client/server pattern (both messages are also terminated with ' '). This is done such that if for example a web client tries to connect, a sensible error message can be returned. Or when a ContextConnection tries to connect to a web server, it will be able to determine the error gracefully. cCs ||_dS)N)r1)r"r;r%r%r&r!szHandShaker.__init__c Cs dt|jtjf}|j}|s>dtfS|jdryT|ddjdd}|d|d}}t|d t|d }}Wn)t k r|j d dt fSYnX|j |} ||krdt fSd ||ffSn|j d dt fSdS)aU _shake_hands_as_host(id) As the host, we wait for the client to ask stuff, so when for example a http client connects, we can stop the connection. Returns (success, info), where info is the id of the context at the other end, or the error message in case success is False. z YOTON!%s.%iFzYOTON!N.r'rrzERROR: could not parse id.TzERROR: this is Yoton.) rget_hexosgetpid_recv_during_handshakingSTOP_HANDSHAKE_TIMEOUTrGsplitintr7_send_during_handshakingSTOP_HANDSHAKE_FAILEDSTOP_HANDSHAKE_SELF) r"idmessageZrequestrSid2_strpid2_strid2pid2rWr%r%r&rs$ "  #     zHandShaker.shake_hands_as_hostc Csdt|jtjf}|j|}|j}|sMdtfS|jdryT|ddjdd}|d|d}}t |d t |d }} Wnt k rdt fSYnX||krdt fSd || ffSn dt fSdS) aU _shake_hands_as_client(id) As the client, we ask the host whether it is a Yoton context and whether the channels we want to support are all right. Returns (success, info), where info is the id of the context at the other end, or the error message in case success is False. z YOTON!%s.%iFzYOTON!rNrr'rrrT) rrrrrrrrGrrr7rr) r"rrrWZresponserSrrrrr%r%r&r\s "  #   z HandShaker.shake_hands_as_clientFcCst|j|d|S)Nz )rr1)r"textr6r%r%r&rsz#HandShaker._send_during_handshakingcCst|jddS)Ng@T)rr1)r"r%r%r&rsz#HandShaker._recv_during_handshakingN) rvrwrxryr!rr\rrr%r%r%r&r[s   ( 'r[c@s:eZdZdZddZddZddZdS) BaseIOThreadzh The base class for the sending and receiving IO threads. Implements some common functionality. cCs6tjj||jd||_|j|_dS)NT)rzr{r!r~r|r1)r"rr%r%r&r!s  zBaseIOThread.__init__c CsK|j}|j}|`|`y|j||Wntk rFYnXdS)z Method to prepare to enter main loop. There is a try-except here to catch exceptions caused by interpreter shutdown. N)r|r1run2r7)r"rr;r%r%r&rs   zBaseIOThread.runc Csd|jj}dd}y,|j||}|rG|j|nWntjk rtt}|jd||fYnTtk rt}t |}|jd||f|d|||YnXdS)z Method to enter main loop. There is a try-except here to catch exceptions in the main loop (such as socket errors and errors due to bugs in the code. zyoton.cSs+tjjt|dtjjdS)N )rE __stderr__writerrk)rer%r%r&writeErr#sz#BaseIOThread.run2..writeErrz%s, %szException in %s.N) __class__rv_runclose_on_problemr=rWSTOP_SOCKET_ERRORrr7STOP_THREAD_ERROR)r"rr;Z classNamerZ stop_reasonmsgZerrmsgr%r%r&rs      zBaseIOThread.run2N)rvrwrxryr!rrr%r%r%r&rs  rc@s"eZdZdZddZdS)r3z The thread that reads packages from the queue and sends them over the socket. It uses a timeout while reading from the queue, so it can send heart beat packages if no packages are send. c Csdt}|j}|j}|j}xotjdy|j|}Wn(|jk rrt}|j sndSYnXx|j D]}||qWq(dS)zI The main loop. Get package from queue, send package to socket. g?rN) rr sendsendallrUrYpopEmptyr rmparts) r"rr;rXZqueueZ socket_sendZsocket_sendallrqpartr%r%r&rHs       zSendingThread._runN)rvrwrxryrr%r%r%r&r3Bs r3c@s:eZdZdZddZddZddZdS) r4a The thread that reads packages from the socket and passes them to the kernel. It uses select() to see if data is available on the socket. This allows using a timeout without putting the socket in timeout mode. If the timeout has expired, the timedout event for the connection is emitted. Upon receiving a package, the _recv_package() method of the context is called, so this thread will eventually dispose the package in one or more queues (of the channel or of another connection). c Cs]|j}|jj}tj}t}d}x,tjdxyt||j }Wn(t k r}t j dt YnX|r|rd}|jj|dnPq=|jsdS|s=d}|jj|dq=q=q=|j|||} | dkrq-nt| tr| Sy|| |Wq-t k rUtdtt Yq-Xq-dS)zN The main loop. Get package from socket, deposit package in queue(s). Frz select(): NTz,Error depositing package in ReceivingThread.)recv_contextZ _recv_packager from_headerrrUrYr_timeoutr7r=rWrtimedoutemitrm _getPackage isinstancerr) r"rr; socket_recvZ recv_packagepackage_from_headerHSZtimedOutr`rqr%r%r&rssB         zReceivingThread._runcCsy|j||}Wntk r.tSYnX||\}}|sKtS|dkr|jdkrin|jdkr|tSdSy|j|||_Wntk rtSYnX|SdS)z< Get exactly one package from the socket. Blocking. rr'N) _recv_n_bytesEOFErrorSTOP_EOFSTOP_LOST_TRACK _source_seqSTOP_CLOSED_FROM_THERE_data)r"rrrheaderrqsizer%r%r&rs$     zReceivingThread._getPackagecCs||}t|dkr*tn|t|8}|dkrJ|S|g}x3|r||}|j||t|8}qVWtj|S)z2 Receive exactly n bytes from the socket. r)lenrrtrjoin)r"rndatarr%r%r&rs       zReceivingThread._recv_n_bytesN)rvrwrxryrrrr%r%r%r&r4es  9 !r4i()1rrErUr=rzyoton yoton.miscrrrrrrrr yoton.corer r r r rrrrryoton.connectionrrrrrrrrrrrrrrrrCrr{rMr[rr3r4r%r%r%r&s8      "(]oP#