pickle length-0 arrays.

IPython/utils/newserialized.py

@@ -1,167 +1,169
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_newserialized -*-
 
 """Refactored serialization classes and interfaces."""
 
 __docformat__ = "restructuredtext en"
 
 # Tell nose to skip this module
 __test__ = {}
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import cPickle as pickle
 
 try:
     import numpy
 except ImportError:
     pass
 
 class SerializationError(Exception):
     pass
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 class ISerialized:
     
     def getData():
         """"""
 
     def getDataSize(units=10.0**6):
         """"""
 
     def getTypeDescriptor():
         """"""
 
     def getMetadata():
         """"""
         
         
 class IUnSerialized:
     
     def getObject():
         """"""
         
 class Serialized(object):
     
     # implements(ISerialized)
     
     def __init__(self, data, typeDescriptor, metadata={}):
         self.data = data
         self.typeDescriptor = typeDescriptor
         self.metadata = metadata
         
     def getData(self):
         return self.data
         
     def getDataSize(self, units=10.0**6):
         return len(self.data)/units
         
     def getTypeDescriptor(self):
         return self.typeDescriptor
         
     def getMetadata(self):
         return self.metadata
 
         
 class UnSerialized(object):
     
     # implements(IUnSerialized)
         
     def __init__(self, obj):
         self.obj = obj
         
     def getObject(self):
         return self.obj
 
         
 class SerializeIt(object):
     
     # implements(ISerialized)
     
     def __init__(self, unSerialized):
         self.data = None
         self.obj = unSerialized.getObject()
         if globals().has_key('numpy') and isinstance(self.obj, numpy.ndarray):
-                if len(self.obj) == 0:         # length 0 arrays can't be reconstructed
-                    raise SerializationError("You cannot send a length 0 array")
+            if len(self.obj.shape) == 0: # length 0 arrays are just pickled
+                self.typeDescriptor = 'pickle'
+                self.metadata = {}
+            else:
                 self.obj = numpy.ascontiguousarray(self.obj, dtype=None)
                 self.typeDescriptor = 'ndarray'
                 self.metadata = {'shape':self.obj.shape,
                                  'dtype':self.obj.dtype.str}
         elif isinstance(self.obj, bytes):
             self.typeDescriptor = 'bytes'
             self.metadata = {}
         elif isinstance(self.obj, buffer):
             self.typeDescriptor = 'buffer'
             self.metadata = {}
         else:
             self.typeDescriptor = 'pickle'
             self.metadata = {}
         self._generateData()
     
     def _generateData(self):
         if self.typeDescriptor == 'ndarray':
             self.data = numpy.getbuffer(self.obj)
         elif self.typeDescriptor in ('bytes', 'buffer'):
             self.data = self.obj
         elif self.typeDescriptor == 'pickle':
             self.data = pickle.dumps(self.obj, -1)
         else:
             raise SerializationError("Really wierd serialization error.")
         del self.obj
         
     def getData(self):
         return self.data
         
     def getDataSize(self, units=10.0**6):
         return 1.0*len(self.data)/units
         
     def getTypeDescriptor(self):
         return self.typeDescriptor
         
     def getMetadata(self):
         return self.metadata
 
 
 class UnSerializeIt(UnSerialized):
     
     # implements(IUnSerialized)
     
     def __init__(self, serialized):
         self.serialized = serialized
         
     def getObject(self):
         typeDescriptor = self.serialized.getTypeDescriptor()
         if globals().has_key('numpy') and typeDescriptor == 'ndarray':
                 buf = self.serialized.getData()
                 if isinstance(buf, (buffer,bytes)):
                     result = numpy.frombuffer(buf, dtype = self.serialized.metadata['dtype'])
                 else:
                     # memoryview
                     result = numpy.array(buf, dtype = self.serialized.metadata['dtype'])
                 result.shape = self.serialized.metadata['shape']
         elif typeDescriptor == 'pickle':
             result = pickle.loads(self.serialized.getData())
         elif typeDescriptor in ('bytes', 'buffer'):
             result = self.serialized.getData()
         else:
             raise SerializationError("Really wierd serialization error.")
         return result
 
 def serialize(obj):
     return SerializeIt(UnSerialized(obj))
     
 def unserialize(serialized):
     return UnSerializeIt(serialized).getObject()

IPython/zmq/parallel/engine.py

@@ -1,139 +1,138
 #!/usr/bin/env python
 """A simple engine that talks to a controller over 0MQ.
 it handles registration, etc. and launches a kernel
 connected to the Controller's Schedulers.
 """
 from __future__ import print_function
 
 import sys
 import time
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # internal
 from IPython.utils.traitlets import Instance, Str, Dict, Int, Type, CFloat
 # from IPython.utils.localinterfaces import LOCALHOST 
 
 from . import heartmonitor
 from .factory import RegistrationFactory
 from .streamkernel import Kernel
 from .streamsession import Message
 from .util import disambiguate_url
 
 class EngineFactory(RegistrationFactory):
     """IPython engine"""
     
     # configurables:
     user_ns=Dict(config=True)
     out_stream_factory=Type('IPython.zmq.iostream.OutStream', config=True)
     display_hook_factory=Type('IPython.zmq.displayhook.DisplayHook', config=True)
     location=Str(config=True)
     timeout=CFloat(2,config=True)
     
     # not configurable:
     id=Int(allow_none=True)
     registrar=Instance('zmq.eventloop.zmqstream.ZMQStream')
     kernel=Instance(Kernel)
     
     
     def __init__(self, **kwargs):
         super(EngineFactory, self).__init__(**kwargs)
         ctx = self.context
         
         reg = ctx.socket(zmq.PAIR)
         reg.setsockopt(zmq.IDENTITY, self.ident)
         reg.connect(self.url)
         self.registrar = zmqstream.ZMQStream(reg, self.loop)
         
     def register(self):
         """send the registration_request"""
         
         self.log.info("registering")
         content = dict(queue=self.ident, heartbeat=self.ident, control=self.ident)
         self.registrar.on_recv(self.complete_registration)
         # print (self.session.key)
         self.session.send(self.registrar, "registration_request",content=content)
     
     def complete_registration(self, msg):
         # print msg
         self._abort_dc.stop()
         ctx = self.context
         loop = self.loop
         identity = self.ident
-        print (identity)
         
         idents,msg = self.session.feed_identities(msg)
         msg = Message(self.session.unpack_message(msg))
         
         if msg.content.status == 'ok':
             self.id = int(msg.content.id)
             
             # create Shell Streams (MUX, Task, etc.):
             queue_addr = msg.content.mux
             shell_addrs = [ str(queue_addr) ]
             task_addr = msg.content.task
             if task_addr:
                 shell_addrs.append(str(task_addr))
             shell_streams = []
             for addr in shell_addrs:
                 stream = zmqstream.ZMQStream(ctx.socket(zmq.PAIR), loop)
                 stream.setsockopt(zmq.IDENTITY, identity)
                 stream.connect(disambiguate_url(addr, self.location))
                 shell_streams.append(stream)
             
             # control stream:
             control_addr = str(msg.content.control)
             control_stream = zmqstream.ZMQStream(ctx.socket(zmq.PAIR), loop)
             control_stream.setsockopt(zmq.IDENTITY, identity)
             control_stream.connect(disambiguate_url(control_addr, self.location))
             
             # create iopub stream:
             iopub_addr = msg.content.iopub
             iopub_stream = zmqstream.ZMQStream(ctx.socket(zmq.PUB), loop)
             iopub_stream.setsockopt(zmq.IDENTITY, identity)
             iopub_stream.connect(disambiguate_url(iopub_addr, self.location))
             
             # launch heartbeat
             hb_addrs = msg.content.heartbeat
             # print (hb_addrs)
             
             # # Redirect input streams and set a display hook.
             if self.out_stream_factory:
                 sys.stdout = self.out_stream_factory(self.session, iopub_stream, u'stdout')
                 sys.stdout.topic = 'engine.%i.stdout'%self.id
                 sys.stderr = self.out_stream_factory(self.session, iopub_stream, u'stderr')
                 sys.stderr.topic = 'engine.%i.stderr'%self.id
             if self.display_hook_factory:
                 sys.displayhook = self.display_hook_factory(self.session, iopub_stream)
                 sys.displayhook.topic = 'engine.%i.pyout'%self.id
             
             self.kernel = Kernel(config=self.config, int_id=self.id, ident=self.ident, session=self.session, 
                     control_stream=control_stream, shell_streams=shell_streams, iopub_stream=iopub_stream, 
                     loop=loop, user_ns = self.user_ns, logname=self.log.name)
             self.kernel.start()
             hb_addrs = [ disambiguate_url(addr, self.location) for addr in hb_addrs ]
             heart = heartmonitor.Heart(*map(str, hb_addrs), heart_id=identity)
             # ioloop.DelayedCallback(heart.start, 1000, self.loop).start()
             heart.start()
             
             
         else:
             self.log.fatal("Registration Failed: %s"%msg)
             raise Exception("Registration Failed: %s"%msg)
         
         self.log.info("Completed registration with id %i"%self.id)
     
     
     def abort(self):
         self.log.fatal("Registration timed out")
         self.session.send(self.registrar, "unregistration_request", content=dict(id=self.id))
         time.sleep(1)
         sys.exit(255)
     
     def start(self):
         dc = ioloop.DelayedCallback(self.register, 0, self.loop)
         dc.start()
         self._abort_dc = ioloop.DelayedCallback(self.abort, self.timeout*1000, self.loop)
         self._abort_dc.start()