def _serve_connection(conn, conn_opts):
    global _g_async_rv_counter
    global _g_server_shutdown
    if isinstance(conn, client.Connection):
        to_client = conn._to_server
        from_client = conn._from_server
    else: # conn is a connected socket
        to_client = conn.makefile("w")
        from_client = conn.makefile("r")
    try:
        peername = conn.getpeername()
    except socket.error:
        peername = ("unknown", "?")
    if opt_debug:
        daemon_log("connected %s:%s" % peername)
    conn_id = "%s-%s" % (timestamp(), id(conn))
    auth_ok = False
    passwords = [k for k in conn_opts.keys() if k.startswith("password.")]
    kill_server_on_close = conn_opts.get("kill-server-on-close", False)
    if passwords:
        # password authentication is required for this connection
        try:
            received_password = pythonshare._recv(from_client)
        except Exception, e:
            daemon_log('error receiving password: %r' % (e,))
            received_password = None
        for password_type in passwords:
            algorithm = password_type.split(".")[1]
            if type(received_password) == str:
                if (algorithm == "plaintext" and
                    received_password == conn_opts[password_type]):
                    auth_ok = True
                elif (hasattr(hashlib, algorithm) and
                      getattr(hashlib, algorithm)(received_password).hexdigest() ==
                      conn_opts[password_type]):
                    auth_ok = True
        try:
            if auth_ok:
                pythonshare._send(messages.Auth_rv(True), to_client)
                if opt_debug:
                    daemon_log("%s:%s authentication ok" % peername)
            elif not received_password is None:
                pythonshare._send(messages.Auth_rv(False), to_client)
                if opt_debug:
                    daemon_log("%s:%s authentication failed" % peername)
        except socket.error:
            daemon_log("authentication failed due to socket error")
            auth_ok = False
    else:
        auth_ok = True # no password required

    whitelist_local = conn_opts.get("whitelist_local", None)

    while auth_ok:
        try:
            obj = pythonshare._recv(from_client)
            if opt_debug:
                daemon_log("%s:%s => %s" % (peername + (obj,)))
        except (EOFError, pythonshare.socket.error):
            break

        if isinstance(obj, messages.Register_ns):
            try:
                _init_remote_namespace(obj.ns, conn, to_client, from_client)
                pythonshare._send(messages.Ns_rv(True), to_client)
                # from this point on, this connection is reserved for
                # sending remote namespace traffic. The connection will be
                # used by other threads, this thread stops here.
                return
            except Exception, e:
                pythonshare._send(messages.Ns_rv(False, exception2string(sys.exc_info())), to_client)

        elif isinstance(obj, messages.Drop_ns):
            try:
                if obj.ns in _g_local_namespaces:
                    _drop_local_namespace(obj.ns)
                elif obj.ns in _g_remote_namespaces:
                    _drop_remote_namespace(obj.ns)
                else:
                    raise ValueError('Unknown namespace "%s"' % (obj.ns,))
                pythonshare._send(messages.Ns_rv(True), to_client)
            except Exception, e:
                if opt_debug:
                    daemon_log("namespace drop error: %s" % (e,))
                pythonshare._send(messages.Ns_rv(False, exception2string(sys.exc_info())), to_client)

        elif isinstance(obj, messages.Request_ns):
            ns = obj.ns
            if (ns in _g_remote_namespaces or
                ns in _g_local_namespaces):
                _register_exported_namespace(ns, conn)
                pythonshare._send(messages.Ns_rv(True), to_client)
                # from this point on, this connection is reserved for
                # receiving executions on requested namespace. This
                # thread starts serving the connection.

        elif isinstance(obj, messages.Exec):
            ns = obj.namespace
            if ns in _g_remote_namespaces: # execute in remote namespace
                try:
                    _fwd_status, _fwd_info = _remote_execute_and_forward(
                        ns, obj, to_client, peername)
                    if _fwd_status == True:
                        # successfully forwarded
                        if opt_debug:
                            daemon_log("%s:%s forwarded %s B" % (peername + (_fwd_info[0],)))
                        exec_rv = None # return value fully forwarded
                    elif _fwd_status == False:
                        # connection to client is broken
                        if opt_debug:
                            daemon_log("%s:%s error after forwarding %s/%s B" % (peername + _fwd_info))
                        break
                    elif _fwd_status is None:
                        # nothing forwarded, send return value by normal means
                        exec_rv = _fwd_info
                except (EOFError, socket.error):
                    daemon_log('connection lost to "%s"' % (ns,))
                    _drop_remote_namespace(ns)
                    break
            else: # execute in local namespace
                if whitelist_local == None or ns in whitelist_local:
                    _init_local_namespace(ns)
                if obj.async:
                    # asynchronous execution, return handle (Async_rv)
                    _g_async_rv_counter += 1
                    rvid = timestamp() + str(_g_async_rv_counter)
                    exec_rv = messages.Async_rv(ns, rvid)
                    _g_async_rvs[ns][rvid] = pythonshare.InProgress()
                    thread.start_new_thread(_local_async_execute, (exec_rv, obj))
                else:
                    # synchronous execution, return true return value
                    exec_rv = _local_execute(obj, conn_id)
            if not exec_rv is None:
                if opt_debug:
                    daemon_log("%s:%s <= %s" % (peername + (exec_rv,)))
                try:
                    try:
                        if obj.recv_cap_data_info():
                            info = pythonshare._send_opt(exec_rv, to_client, obj.recv_caps)
                            if info:
                                sent_info = " %s B, format:%s" % (
                                    info.data_length, info.data_format)
                            else:
                                sent_info = ""
                        else:
                            pythonshare._send(exec_rv, to_client)
                            sent_info = ""
                        if opt_debug:
                            daemon_log("%s:%s sent%s" % (peername + (sent_info,)))
                    except (EOFError, socket.error):
                        break
                except (TypeError, ValueError, cPickle.PicklingError): # pickling rv fails
                    exec_rv.expr_rv = messages.Unpicklable(exec_rv.expr_rv)
                    try:
                        pythonshare._send(exec_rv, to_client)
                    except (EOFError, socket.error):
                        break

        elif isinstance(obj, messages.Server_ctl):
            if obj.command == "die":
                ns = obj.args[0]
                if ns in _g_remote_namespaces:
                    try:
                        rv = _remote_execute(ns, obj)
                        if opt_debug:
                            daemon_log("%s:%s <= %s" % (peername + (rv,)))
                        pythonshare._send(rv, to_client)
                    except (EOFError, socket.error): # connection lost
                        daemon_log('connection lost to "%s"' % (ns,))
                        _drop_remote_namespace(ns)
                        break
                else:
                    _g_server_shutdown = True
                    server_ctl_rv = messages.Server_ctl_rv(0, "shutting down")
                    pythonshare._send(server_ctl_rv, to_client)
                    if _g_wake_server_function:
                        _g_wake_server_function()
                    break
            elif obj.command == "unlock":
                try:
                    ns = obj.args[0]
                    if ns in _g_remote_namespaces:
                        try:
                            rv = _remote_execute(ns, obj)
                        except (EOFError, socket.error): # connection lost
                            daemon_log('connection lost to "%s"' % (ns,))
                            _drop_remote_namespace(ns)
                            break
                    elif ns in _g_local_namespace_locks:
                        try:
                            _g_local_namespace_locks[ns].release()
                            server_ctl_rv = messages.Server_ctl_rv(
                                0, "%s unlocked" % (repr(ns),))
                        except thread.error, e:
                            server_ctl_rv = messages.Server_ctl_rv(
                                1, "%s already unlocked" %
                                (repr(ns),))
                    elif ns in _g_local_namespaces:
                        server_ctl_rv = messages.Server_ctl_rv(
                            2, "namespace %s is not locked" % (repr(ns),))
                    else:
                        server_ctl_rv = messages.Server_ctl_rv(
                            -1, "unknown namespace %s" % (repr(ns),))
                    if opt_debug:
                        daemon_log("%s:%s <= %s" % (peername + (server_ctl_rv,)))
                    pythonshare._send(server_ctl_rv, to_client)
                except Exception, e:
                    if opt_debug:
                        daemon_log("Exception in handling %s: %s" % (obj, e))
        else:
            daemon_log("unknown message type: %s in %s" % (type(obj), obj))
            pythonshare._send(messages.Auth_rv(False), to_client)
            auth_ok = False
    if opt_debug:
        daemon_log("disconnected %s:%s" % peername)
    _connection_lost(conn_id, to_client, from_client, conn)
    if kill_server_on_close:
        _g_server_shutdown = True
        if _g_wake_server_function:
            _g_wake_server_function()