home *** CD-ROM | disk | FTP | other *** search
/ Chip 2011 November / CHIP_2011_11.iso / Programy / Narzedzia / Inkscape / Inkscape-0.48.2-1-win32.exe / python / Lib / multiprocessing / reduction.py < prev    next >
Encoding:
Python Source  |  2010-05-29  |  5.2 KB  |  190 lines
  1. #
  2. # Module to allow connection and socket objects to be transferred
  3. # between processes
  4. #
  5. # multiprocessing/reduction.py
  6. #
  7. # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
  8. #
  9.  
  10. __all__ = []
  11.  
  12. import os
  13. import sys
  14. import socket
  15. import threading
  16.  
  17. import _multiprocessing
  18. from multiprocessing import current_process
  19. from multiprocessing.forking import Popen, duplicate, close, ForkingPickler
  20. from multiprocessing.util import register_after_fork, debug, sub_debug
  21. from multiprocessing.connection import Client, Listener
  22.  
  23.  
  24. #
  25. #
  26. #
  27.  
  28. if not(sys.platform == 'win32' or hasattr(_multiprocessing, 'recvfd')):
  29.     raise ImportError('pickling of connections not supported')
  30.  
  31. #
  32. # Platform specific definitions
  33. #
  34.  
  35. if sys.platform == 'win32':
  36.     import _subprocess
  37.     from ._multiprocessing import win32
  38.  
  39.     def send_handle(conn, handle, destination_pid):
  40.         process_handle = win32.OpenProcess(
  41.             win32.PROCESS_ALL_ACCESS, False, destination_pid
  42.             )
  43.         try:
  44.             new_handle = duplicate(handle, process_handle)
  45.             conn.send(new_handle)
  46.         finally:
  47.             close(process_handle)
  48.  
  49.     def recv_handle(conn):
  50.         return conn.recv()
  51.  
  52. else:
  53.     def send_handle(conn, handle, destination_pid):
  54.         _multiprocessing.sendfd(conn.fileno(), handle)
  55.  
  56.     def recv_handle(conn):
  57.         return _multiprocessing.recvfd(conn.fileno())
  58.  
  59. #
  60. # Support for a per-process server thread which caches pickled handles
  61. #
  62.  
  63. _cache = set()
  64.  
  65. def _reset(obj):
  66.     global _lock, _listener, _cache
  67.     for h in _cache:
  68.         close(h)
  69.     _cache.clear()
  70.     _lock = threading.Lock()
  71.     _listener = None
  72.  
  73. _reset(None)
  74. register_after_fork(_reset, _reset)
  75.  
  76. def _get_listener():
  77.     global _listener
  78.  
  79.     if _listener is None:
  80.         _lock.acquire()
  81.         try:
  82.             if _listener is None:
  83.                 debug('starting listener and thread for sending handles')
  84.                 _listener = Listener(authkey=current_process().authkey)
  85.                 t = threading.Thread(target=_serve)
  86.                 t.daemon = True
  87.                 t.start()
  88.         finally:
  89.             _lock.release()
  90.  
  91.     return _listener
  92.  
  93. def _serve():
  94.     from .util import is_exiting, sub_warning
  95.  
  96.     while 1:
  97.         try:
  98.             conn = _listener.accept()
  99.             handle_wanted, destination_pid = conn.recv()
  100.             _cache.remove(handle_wanted)
  101.             send_handle(conn, handle_wanted, destination_pid)
  102.             close(handle_wanted)
  103.             conn.close()
  104.         except:
  105.             if not is_exiting():
  106.                 import traceback
  107.                 sub_warning(
  108.                     'thread for sharing handles raised exception :\n' +
  109.                     '-'*79 + '\n' + traceback.format_exc() + '-'*79
  110.                     )
  111.  
  112. #
  113. # Functions to be used for pickling/unpickling objects with handles
  114. #
  115.  
  116. def reduce_handle(handle):
  117.     if Popen.thread_is_spawning():
  118.         return (None, Popen.duplicate_for_child(handle), True)
  119.     dup_handle = duplicate(handle)
  120.     _cache.add(dup_handle)
  121.     sub_debug('reducing handle %d', handle)
  122.     return (_get_listener().address, dup_handle, False)
  123.  
  124. def rebuild_handle(pickled_data):
  125.     address, handle, inherited = pickled_data
  126.     if inherited:
  127.         return handle
  128.     sub_debug('rebuilding handle %d', handle)
  129.     conn = Client(address, authkey=current_process().authkey)
  130.     conn.send((handle, os.getpid()))
  131.     new_handle = recv_handle(conn)
  132.     conn.close()
  133.     return new_handle
  134.  
  135. #
  136. # Register `_multiprocessing.Connection` with `ForkingPickler`
  137. #
  138.  
  139. def reduce_connection(conn):
  140.     rh = reduce_handle(conn.fileno())
  141.     return rebuild_connection, (rh, conn.readable, conn.writable)
  142.  
  143. def rebuild_connection(reduced_handle, readable, writable):
  144.     handle = rebuild_handle(reduced_handle)
  145.     return _multiprocessing.Connection(
  146.         handle, readable=readable, writable=writable
  147.         )
  148.  
  149. ForkingPickler.register(_multiprocessing.Connection, reduce_connection)
  150.  
  151. #
  152. # Register `socket.socket` with `ForkingPickler`
  153. #
  154.  
  155. def fromfd(fd, family, type_, proto=0):
  156.     s = socket.fromfd(fd, family, type_, proto)
  157.     if s.__class__ is not socket.socket:
  158.         s = socket.socket(_sock=s)
  159.     return s
  160.  
  161. def reduce_socket(s):
  162.     reduced_handle = reduce_handle(s.fileno())
  163.     return rebuild_socket, (reduced_handle, s.family, s.type, s.proto)
  164.  
  165. def rebuild_socket(reduced_handle, family, type_, proto):
  166.     fd = rebuild_handle(reduced_handle)
  167.     _sock = fromfd(fd, family, type_, proto)
  168.     close(fd)
  169.     return _sock
  170.  
  171. ForkingPickler.register(socket.socket, reduce_socket)
  172.  
  173. #
  174. # Register `_multiprocessing.PipeConnection` with `ForkingPickler`
  175. #
  176.  
  177. if sys.platform == 'win32':
  178.  
  179.     def reduce_pipe_connection(conn):
  180.         rh = reduce_handle(conn.fileno())
  181.         return rebuild_pipe_connection, (rh, conn.readable, conn.writable)
  182.  
  183.     def rebuild_pipe_connection(reduced_handle, readable, writable):
  184.         handle = rebuild_handle(reduced_handle)
  185.         return _multiprocessing.PipeConnection(
  186.             handle, readable=readable, writable=writable
  187.             )
  188.  
  189.     ForkingPickler.register(_multiprocessing.PipeConnection, reduce_pipe_connection)
  190.