Pool to Celery group

Question

I'm using Celery to launch a task, this task includes a port scanner which is built using multiprocessing and ThreadPool. When I launch it I get the following error, looking for an alternative or tutorial to migrate my existing code:

[2015-10-25 18:53:57,090: ERROR/Worker-3] daemonic processes are not allowed to have children
Traceback (most recent call last):
  File "/Users/user/Documents/OpenSource/Development/Python/test/utils/port_scanner.py", line 57, in is_port_opened
    ports = list(scan_ports(server=server, port=int(port)))
  File "/Users/user/Documents/OpenSource/Development/Python/test/utils/port_scanner.py", line 37, in scan_ports
    p = Pool(NUM_CORES)
  File "/usr/local/Cellar/python/2.7.10_2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/__init__.py", line 232, in Pool
    return Pool(processes, initializer, initargs, maxtasksperchild)
  File "/usr/local/Cellar/python/2.7.10_2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 159, in __init__
    self._repopulate_pool()
  File "/usr/local/Cellar/python/2.7.10_2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 223, in _repopulate_pool
    w.start()
  File "/usr/local/Cellar/python/2.7.10_2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/process.py", line 124, in start
    'daemonic processes are not allowed to have children'
AssertionError: daemonic processes are not allowed to have children

This is my code:

import socket
from functools import partial
from multiprocessing import Pool
from multiprocessing.pool import ThreadPool
from errno import ECONNREFUSED

NUM_CORES = 4
def portscan(target, port):
    try:
        # Create Socket
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(12)
        s.connect((target, port))
        print 'port_scanner.is_port_opened() ' + str(port) + " is opened"
        return port
    except socket.error as err:
        print str(err)
        if err.errno == ECONNREFUSED:
            return False


# Wrapper function that calls portscanner
def scan_ports(server=None, port=None, portStart=None, portEnd=None, **kwargs):
    p = Pool(NUM_CORES)
    ping_host = partial(portscan, server)

    if portStart and portStart:
        # Filter returns Booleans
        return filter(bool, p.map(ping_host, range(portStart, portStart)))
    else:
        return filter(bool, p.map(ping_host, range(port, port + 1)))


# Check if port is opened
def is_port_opened(server=None, port=None, **kwargs):
    try:
        # check if its IP Address:
        ports = list(scan_ports(server=server, port=int(port)))
        if len(ports) != 0:
            print 'port_scanner.is_port_opened() ' + str(len(ports)) + " port(s) available."
            return True
        else:
            print 'port_scanner.is_port_opened() port not opened: (' + str(port) + ')'
            return False

    except Exception, e:
        print str(e)

if __name__ == '__main__':
    is_port_opened(server='110.10.0.144', port=22)

score 1 · Accepted Answer · edited May 23 '17 at 10:27

Your problem is pretty close to the one experienced in this question. Daemonic processes cannot spawn child processes. The reason behind it, is because a daemonic process gets terminated forcefully by its parent. But if the parent would be a daemonic process as well, it would be terminated forcefully without having time to do the same with its child.

multiprocessing.Pool uses daemonic processes to ensure they don't leak when your program extits. I assume Celery does the same to run your task. Therefore you're spawning daemonic processes within a daemon process.

A possible solution would be to quickly implement a Pool of processes yourself using Process and Queue objects from multiprocessing library. You would spawn non daemonic processes and feed them through a queue.

Pool to Celery group

1 Answers1