title: Pickle error when submitting task using dask
body: <p>I am trying to execute a simple task(an instance method) using dask(async) framework but it fails with serialization error.</p>
  
  <p>Can someone point me in right direction.</p>
  
  <p>Here is the code that I am running:</p>
  
  <pre><code>from dask.distributed import Client, as_completed
  import time
  
  class DaskConnect:
  
  def __init__(self):
      print("Initialized:",self.__class__.__name__)
      self.scheduler_host="192.168.0.4"
      self.scheduler_port="8786"
  
  def connect(self):
      self.client = Client(self.scheduler_host+":"+self.scheduler_port)
      # self.client = Client()
      return self.client
  
  def disconnect(self):
      self.client.close()
  
  class TestDask:
  def __init__(self):
      print("Initialized:",self.__class__.__name__)
      self.dask_client=DaskConnect().connect()
  
  def do_task(self,msg):
      time.sleep(30)
      return msg
  
  def run(self):
      tasks=[1]
      # tasks = [1, 2, 3, 4, 5]
      futures=[]
      for task in tasks:
          print("Submitting:",task)
          future = self.dask_client.submit(self.do_task, "Task:"+str(task))
          futures.append(future)
  
      for future in as_completed(futures):
          result = future.result()
          print("Result",result)
  
  TestDask().run()
  </code></pre>
  
  <p>Error:</p>
  
  <blockquote>
    <p>distributed.protocol.pickle - INFO - Failed to serialize main.TestDask object at 0x101c408d0>>.
    Exception: can't pickle select.kqueue objects Traceback (most recent
    call last):</p>
  </blockquote>