| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218 |
- #
- # Module to support the pickling of different types of connection
- # objects and file objects so that they can be transferred between
- # different processes.
- #
- # processing/reduction.py
- #
- # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
- #
- __all__ = []
- import os
- import sys
- import socket
- import threading
- import copy_reg
- import processing
- from processing import _processing
- from processing.logger import debug, subDebug, subWarning
- from processing.forking import thisThreadIsSpawning
- from processing.process import _registerAfterFork
- #
- #
- #
- connections_are_picklable = (
- sys.platform == 'win32' or hasattr(_processing, 'recvFd')
- )
- try:
- fromfd = socket.fromfd
- except AttributeError:
- def fromfd(fd, family, type, proto=0):
- s = socket._socket.socket()
- _processing.changeFd(s, fd, family, type, proto)
- return s
- #
- # Platform specific definitions
- #
- if sys.platform == 'win32':
- import _subprocess
- from processing._processing import win32
- closeHandle = win32.CloseHandle
-
- def duplicateHandle(handle):
- return _subprocess.DuplicateHandle(
- _subprocess.GetCurrentProcess(), handle,
- _subprocess.GetCurrentProcess(),
- 0, False, _subprocess.DUPLICATE_SAME_ACCESS
- ).Detach()
- def sendHandle(conn, handle, destination_pid):
- process_handle = win32.OpenProcess(
- win32.PROCESS_ALL_ACCESS, False, destination_pid
- )
- try:
- new_handle = _subprocess.DuplicateHandle(
- _subprocess.GetCurrentProcess(), handle,
- process_handle, 0, False, _subprocess.DUPLICATE_SAME_ACCESS
- )
- conn.send(new_handle.Detach())
- finally:
- win32.CloseHandle(process_handle)
-
- def recvHandle(conn):
- return conn.recv()
- def isInheritableHandle(handle):
- return (win32.GetHandleInformation(handle) & win32.HANDLE_FLAG_INHERIT)
- else:
- closeHandle = os.close
- duplicateHandle = os.dup
-
- def sendHandle(conn, handle, destination_pid):
- _processing.sendFd(conn.fileno(), handle)
-
- def recvHandle(conn):
- return _processing.recvFd(conn.fileno())
-
- def isInheritableHandle(handle):
- return True
- #
- # Support for a per-process server thread which caches pickled handles
- #
- _cache = set()
- def _reset(obj):
- global _lock, _listener, _cache
- for h in _cache:
- closeHandle(h)
- _cache.clear()
- _lock = threading.Lock()
- _listener = None
- _reset(None)
- _registerAfterFork(_reset, _reset)
- def _getListener():
- global _listener
- if _listener is None:
- _lock.acquire()
- try:
- if _listener is None:
- from processing.connection import Listener
- debug('starting listener and thread for sending handles')
- _listener = Listener(authenticate=True)
- t = threading.Thread(target=_serve)
- t.setDaemon(True)
- t.start()
- finally:
- _lock.release()
- return _listener
- def _serve():
- while 1:
- try:
- conn = _listener.accept()
- handle_wanted, destination_pid = conn.recv()
- _cache.remove(handle_wanted)
- sendHandle(conn, handle_wanted, destination_pid)
- closeHandle(handle_wanted)
- conn.close()
- except (SystemExit, KeyboardInterrupt):
- raise
- except:
- if not processing.currentProcess()._exiting:
- import traceback
- subWarning(
- 'thread for sharing handles raised exception :\n' +
- '-'*79 + '\n' + traceback.format_exc() + '-'*79
- )
-
- #
- # Functions to be used for pickling/unpickling objects with handles
- #
- def reduceHandle(handle):
- if thisThreadIsSpawning() and isInheritableHandle(handle):
- return (None, handle, True)
- dup_handle = duplicateHandle(handle)
- _cache.add(dup_handle)
- subDebug('reducing handle %d', handle)
- return (_getListener().address, dup_handle, False)
- def rebuildHandle(pickled_data):
- from processing.connection import Client
- address, handle, inherited = pickled_data
- if inherited:
- return handle
- subDebug('rebuilding handle %d', handle)
- conn = Client(address, authenticate=True)
- conn.send((handle, os.getpid()))
- new_handle = recvHandle(conn)
- conn.close()
- return new_handle
- #
- # Register `_processing.Connection` with `copy_reg`
- #
- def reduceConnection(conn):
- return rebuildConnection, (reduceHandle(conn.fileno()),)
- def rebuildConnection(reduced_handle):
- fd = rebuildHandle(reduced_handle)
- return _processing.Connection(fd, duplicate=False)
- copy_reg.pickle(_processing.Connection, reduceConnection)
- #
- # Register `socket.socket` with `copy_reg`
- #
- def reduceSocket(s):
- try:
- Family, Type, Proto = s.family, s.type, s.proto
- except AttributeError:
- # have to guess family, type, proto
- address = s.getsockname()
- Family = type(address) is str and socket.AF_UNIX or socket.AF_INET
- Type = s.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE)
- Proto = 0
- reduced_handle = reduceHandle(s.fileno())
- return rebuildSocket, (reduced_handle, Family, Type, Proto)
- def rebuildSocket(reduced_handle, family, type, proto):
- fd = rebuildHandle(reduced_handle)
- _sock = fromfd(fd, family, type, proto)
- closeHandle(fd)
- return socket.socket(_sock=_sock)
- copy_reg.pickle(socket.socket, reduceSocket)
- #
- # Register `_processing.PipeConnection` with `copy_reg`
- #
- if sys.platform == 'win32':
-
- def reducePipeConnection(conn):
- return rebuildPipeConnection, (reduceHandle(conn.fileno()),)
-
- def rebuildPipeConnection(reduced_handle):
- handle = rebuildHandle(reduced_handle)
- return _processing.PipeConnection(handle, duplicate=False)
-
- copy_reg.pickle(_processing.PipeConnection, reducePipeConnection)
|