static bool
maybe_populate_impl(
        const char* name,
        const char* description,
              T&    destination,
        const T&    source,
        const bool  verbose,
        bool (* const default_value)(const T& v))
{
    if (default_value(destination)) {
        if (verbose) {
            // Populating a default isn't interesting, hence DEBUG0
            if (description) {
                DEBUG0("Populating " << name
                      << " (" << description << ") to be " << source);
            } else {
                DEBUG0("Populating " << name << " to be " << source);
            }
        }
        destination = source;
        return true;
    }

    if (verbose) {
        if (description) {
            // Retaining an existing setting is interesting, hence INFO0
            INFO0("Clutching onto " << name
                   << " (" << description << ") of " << destination);
        } else {
            INFO0("Clutching onto " << name << " of " << destination);
        }
    }
    return false;
}

void connection_accept_loop(void)
{
    connection_t *con;

    _build_pool();

    while (global.running == ICE_RUNNING)
    {
        if (global . schedule_config_reread)
        {
            /* reread config file */
            INFO0("Scheduling config reread ...");

            connection_inject_event(EVENT_CONFIG_READ, NULL);
            global . schedule_config_reread = 0;
        }

        con = _accept_connection();

        if (con) {
            _add_connection(con);
        }
    }

    /* Give all the other threads notification to shut down */
    thread_cond_broadcast(&global.shutdown_cond);

    _destroy_pool();

    /* wait for all the sources to shutdown */
    thread_rwlock_wlock(&_source_shutdown_rwlock);
    thread_rwlock_unlock(&_source_shutdown_rwlock);
}

/** Receiver function  */
int slave(int argc, char *argv[])
{
  m_task_t task = NULL;
  int res;
  int id = -1;
  char mailbox[80];

  xbt_assert1(sscanf(argv[1],"%d", &id),
	 "Invalid argument %s\n",argv[1]);

  sprintf(mailbox,"slave-%d",id);

  while(1) {
    res = MSG_task_receive(&(task), mailbox);
    xbt_assert0(res == MSG_OK, "MSG_task_get failed");

//  INFO1("Received \"%s\"", MSG_task_get_name(task));
    if (!strcmp(MSG_task_get_name(task),"finalize")) {
	MSG_task_destroy(task);
	break;
    }

//    INFO1("Processing \"%s\"", MSG_task_get_name(task));
    MSG_task_execute(task);
//    INFO1("\"%s\" done", MSG_task_get_name(task));
    MSG_task_destroy(task);
    task = NULL;
  }
  INFO0("I'm done. See you!");
  return 0;
} /* end_of_slave */

ogg_codec_t *initial_speex_page (format_plugin_t *plugin, ogg_page *page)
{
    ogg_state_t *ogg_info = plugin->_state;
    ogg_codec_t *codec = acalloc (1, sizeof (ogg_codec_t));
    ogg_packet packet;
    SpeexHeader *header;

    ogg_stream_init (&codec->os, ogg_page_serialno (page));
    ogg_stream_pagein (&codec->os, page);

    ogg_stream_packetout (&codec->os, &packet);

    DEBUG0("checking for speex codec");
    header = speex_packet_to_header ((char*)packet.packet, packet.bytes);
    if (header == NULL)
    {
        ogg_stream_clear (&codec->os);
        free (header);
        free (codec);
        return NULL;
    }
    INFO0 ("seen initial speex header");
    codec->process_page = process_speex_page;
    codec->codec_free = speex_codec_free;
    codec->headers = 1;
    format_ogg_attach_header (ogg_info, page);
    free (header);
    return codec;
}

void fserve_shutdown(void)
{
    thread_spin_lock (&pending_lock);
    run_fserv = 0;
    while (pending_list)
    {
        fserve_t *to_go = (fserve_t *)pending_list;
        pending_list = to_go->next;

        fserve_client_destroy (to_go);
    }
    while (active_list)
    {
        fserve_t *to_go = active_list;
        active_list = to_go->next;
        fserve_client_destroy (to_go);
    }

    if (mimetypes)
        avl_tree_free (mimetypes, _delete_mapping);

    thread_spin_unlock (&pending_lock);
    thread_spin_destroy (&pending_lock);
    INFO0("file serving stopped");
}

static void *log_commit_thread (void *arg)
{
   INFO0 ("started");
   while (1)
   {
       int ret = util_timed_wait_for_fd (logger_fd[0], 5000);
       if (ret == 0) continue;
       if (ret > 0)
       {
           char cm[80];
           ret = pipe_read (logger_fd[0], cm, sizeof cm);
           if (ret > 0)
           {
               // fprintf (stderr, "logger woken with %d\n", ret);
               log_commit_entries ();
               continue;
           }
       }
       if (ret < 0 && sock_recoverable (sock_error()))
           continue;
       int err = sock_error();
       sock_close (logger_fd[0]);
       sock_close (logger_fd[1]);
       if (worker_count)
       {
           worker_control_create (logger_fd);
           ERROR1 ("logger received code %d", err);
           continue;
       }
       // fprintf (stderr, "logger closed with zero workers\n");
       break;
   }
   return NULL;
}

static void worker_start (void)
{
    worker_t *handler = calloc (1, sizeof(worker_t));

    worker_control_create (&handler->wakeup_fd[0]);

    handler->pending_clients_tail = &handler->pending_clients;
    thread_spin_create (&handler->lock);
    handler->last_p = &handler->clients;

    thread_rwlock_wlock (&workers_lock);
    if (worker_incoming == NULL)
    {
        worker_incoming = handler;
        handler->thread = thread_create ("worker", worker, handler, THREAD_ATTACHED);
        thread_rwlock_unlock (&workers_lock);
        INFO0 ("starting incoming worker thread");
        worker_start();  // single level recursion, just get a special worker thread set up
        return;
    }
    handler->next = workers;
    workers = handler;
    worker_count++;
    worker_least_used = worker_balance_to_check = workers;
    thread_rwlock_unlock (&workers_lock);

    handler->thread = thread_create ("worker", worker, handler, THREAD_ATTACHED);
}

/* This is called with the config lock held */
static void *alloc_thread_data (auth_t *auth)
{
    auth_thread_data *atd = calloc (1, sizeof (auth_thread_data));
    ice_config_t *config = config_get_config_unlocked();
    auth_url *url = auth->state;
    atd->server_id = strdup (config->server_id);

    atd->curl = curl_easy_init ();
    curl_easy_setopt (atd->curl, CURLOPT_USERAGENT, atd->server_id);
    curl_easy_setopt (atd->curl, CURLOPT_HEADERFUNCTION, handle_returned_header);
    curl_easy_setopt (atd->curl, CURLOPT_WRITEFUNCTION, handle_returned_data);
    curl_easy_setopt (atd->curl, CURLOPT_NOSIGNAL, 1L);
    curl_easy_setopt (atd->curl, CURLOPT_TIMEOUT, (long)url->timeout);
#ifdef CURLOPT_PASSWDFUNCTION
    curl_easy_setopt (atd->curl, CURLOPT_PASSWDFUNCTION, my_getpass);
#endif
    curl_easy_setopt (atd->curl, CURLOPT_ERRORBUFFER, &atd->errormsg[0]);
    curl_easy_setopt (atd->curl, CURLOPT_FOLLOWLOCATION, 1);
#ifdef CURLOPT_POSTREDIR
    curl_easy_setopt (atd->curl, CURLOPT_POSTREDIR, CURL_REDIR_POST_ALL);
#endif
    if (auth->flags & AUTH_SKIP_IF_SLOW)
        curl_easy_setopt (atd->curl, CURLOPT_SSL_VERIFYPEER, 0L);
    INFO0 ("...handler data initialized");
    return atd;
}

ogg_codec_t *initial_opus_page (format_plugin_t *plugin, ogg_page *page)
{
    ogg_state_t *ogg_info = plugin->_state;
    ogg_codec_t *codec = calloc (1, sizeof (ogg_codec_t));
    ogg_packet packet;

    ogg_stream_init (&codec->os, ogg_page_serialno (page));
    ogg_stream_pagein (&codec->os, page);

    ogg_stream_packetout (&codec->os, &packet);

    DEBUG0("checking for opus codec");
    if (strncmp((char *)packet.packet, "OpusHead", 8) != 0)
    {
        ogg_stream_clear (&codec->os);
        free (codec);
        return NULL;
    }
    INFO0 ("seen initial opus header");
    codec->process_page = process_opus_page;
    codec->codec_free = opus_codec_free;
    codec->headers = 1;
    codec->parent = ogg_info;
    codec->name = "Opus";
    format_ogg_attach_header (codec, page);
    return codec;
}

/** Receiver function  */
int slave(int argc, char *argv[])
{
  m_task_t task = NULL;
  int res;
  int id = -1;

  xbt_assert1(sscanf(argv[1],"%d", &id),
	 "Invalid argument %s\n",argv[1]);

  MSG_process_sleep(1); /* Make sure the master is done creating the mailboxes */
   
  while(1) {
    res = MSG_mailbox_get_task_ext(mb[id], &(task), NULL, -1);
    xbt_assert0(res == MSG_OK, "MSG_task_get failed");

    INFO1("Received \"%s\"", MSG_task_get_name(task));
    if (!strcmp(MSG_task_get_name(task),"finalize")) {
	MSG_task_destroy(task);
	break;
    }
     
    INFO1("Processing \"%s\"", MSG_task_get_name(task));
    MSG_task_execute(task);
    INFO1("\"%s\" done", MSG_task_get_name(task));
    MSG_task_destroy(task);
    task = NULL;
  }
  INFO0("I'm done. See you!");
  return 0;
} /* end_of_slave */

void fserve_initialize(void)
{
    if (fserve_running) return;

    ice_config_t *config = config_get_config();

    mimetypes = NULL;
    thread_spin_create (&pending_lock);
#ifndef HAVE_PREAD
    thread_mutex_create (&seekread_lock);
#endif
    fh_cache = avl_tree_new (_compare_fh, NULL);

    fserve_recheck_mime_types (config);
    config_release_config();

    stats_event_flags (NULL, "file_connections", "0", STATS_COUNTERS);
    fserve_running = 1;
    memset (&no_file, 0, sizeof (no_file));
    thread_mutex_create (&no_file.lock);
    no_file.clients = avl_tree_new (client_compare, NULL);
    no_file.refcount = 1;
    no_file.expire = (time_t)-1;
    no_file.f = -1;
    avl_insert (fh_cache, &no_file);
    INFO0("file serving started");
}

void fserve_shutdown(void)
{
    fserve_running = 0;
    if (mimetypes)
        avl_tree_free (mimetypes, _delete_mapping);
    if (fh_cache)
    {
        int count = 20;
        avl_delete (fh_cache, &no_file, NULL);
        while (fh_cache->length > 1 && count)
        {
            fh_node *fh = fh_cache->root->right->key;
            if (fh && fh->refcount == 0)
            {
                remove_fh_from_cache (fh);
                continue;
            }
            DEBUG1 ("waiting for %u entries to clear", fh_cache->length);
            thread_sleep (100000);
            count--;
        }
        avl_tree_free (fh_cache, _delete_fh);
    }

    thread_spin_destroy (&pending_lock);
#ifndef HAVE_PREAD
    thread_mutex_destroy (&seekread_lock);
#endif
    INFO0("file serving stopped");
}

void *stats_connection(void *arg)
{
    client_t *client = (client_t *)arg;
    stats_event_t *event;
    event_listener_t listener;

    INFO0 ("stats client starting");

    event_queue_init (&listener.queue);
    /* increment the thread count */
    thread_mutex_lock(&_stats_mutex);
    _stats_threads++;
    stats_event_args (NULL, "stats", "%d", _stats_threads);
    thread_mutex_unlock(&_stats_mutex);

    thread_mutex_create (&(listener.mutex));

    _register_listener (&listener);

    while (_stats_running) {
        thread_mutex_lock (&listener.mutex);
        event = _get_event_from_queue (&listener.queue);
        thread_mutex_unlock (&listener.mutex);
        if (event != NULL) {
            if (_send_event_to_client(event, client) < 0) {
                _free_event(event);
                break;
            }
            _free_event(event);
            continue;
        }
        thread_sleep (500000);
    }

    thread_mutex_lock(&_stats_mutex);
    _unregister_listener (&listener);
    _stats_threads--;
    stats_event_args (NULL, "stats", "%d", _stats_threads);
    thread_mutex_unlock(&_stats_mutex);

    thread_mutex_destroy (&listener.mutex);
    client_destroy (client);
    INFO0 ("stats client finished");

    return NULL;
}

void yp_shutdown ()
{
    yp_running = 0;
    yp_update = 1;
    if (yp_thread)
        thread_join (yp_thread);
    curl_global_cleanup();
    INFO0 ("YP thread down");
}

static http_parser_t *get_relay_response (connection_t *con, const char *mount,
        const char *server, int ask_for_metadata, const char *auth_header)
{
    ice_config_t *config = config_get_config ();
    char *server_id = strdup (config->server_id);
    http_parser_t *parser = NULL;
    char response [4096];

    config_release_config ();

    /* At this point we may not know if we are relaying an mp3 or vorbis
     * stream, but only send the icy-metadata header if the relay details
     * state so (the typical case).  It's harmless in the vorbis case. If
     * we don't send in this header then relay will not have mp3 metadata.
     */
    sock_write (con->sock, "GET %s HTTP/1.0\r\n"
            "User-Agent: %s\r\n"
            "Host: %s\r\n"
            "%s"
            "%s"
            "\r\n",
            mount,
            server_id,
            server,
            ask_for_metadata ? "Icy-MetaData: 1\r\n" : "",
            auth_header ? auth_header : "");

    free (server_id);
    memset (response, 0, sizeof(response));
    if (util_read_header (con->sock, response, 4096, READ_ENTIRE_HEADER) == 0)
    {
        INFO0 ("Header read failure");
        return NULL;
    }
    parser = httpp_create_parser();
    httpp_initialize (parser, NULL);
    if (! httpp_parse_response (parser, response, strlen(response), mount))
    {
        INFO0 ("problem parsing response from relay");
        httpp_destroy (parser);
        return NULL;
    }
    return parser;
}

void fserve_shutdown(void)
{
    if(!run_fserv)
        return;

    run_fserv = 0;
    thread_join(fserv_thread);
    INFO0("file serving thread stopped");
    avl_tree_free(mimetypes, _delete_mapping);
}

void auth_shutdown (void)
{
    if (allow_auth == 0)
        return;
    allow_auth = 0;
    thread_rwlock_wlock (&auth_lock);
    thread_rwlock_unlock (&auth_lock);
    thread_rwlock_destroy (&auth_lock);
    INFO0 ("Auth shutdown complete");
}

/* check if the provided BOS page is the start of a vorbis stream. If so
 * then setup a structure so it can be used
 */
ogg_codec_t *initial_vorbis_page (format_plugin_t *plugin, ogg_page *page)
{
    ogg_state_t *ogg_info = plugin->_state;
    ogg_codec_t *codec = calloc (1, sizeof (ogg_codec_t));
    ogg_packet packet;

    vorbis_codec_t *vorbis = calloc (1, sizeof (vorbis_codec_t));

    ogg_stream_init (&codec->os, ogg_page_serialno (page));
    ogg_stream_pagein (&codec->os, page);

    vorbis_info_init (&vorbis->vi);
    vorbis_comment_init (&vorbis->vc);

    ogg_stream_packetout (&codec->os, &packet);

    DEBUG0("checking for vorbis codec");
    if (vorbis_synthesis_headerin (&vorbis->vi, &vorbis->vc, &packet) < 0)
    {
        ogg_stream_clear (&codec->os);
        vorbis_info_clear (&vorbis->vi);
        vorbis_comment_clear (&vorbis->vc);
        free (vorbis);
        free (codec);
        return NULL;
    }
    INFO0 ("seen initial vorbis header");
    codec->specific = vorbis;
    codec->codec_free = vorbis_codec_free;
    codec->headers = 1;
    codec->parent = ogg_info;
    codec->name = "Vorbis";

    free_ogg_packet (vorbis->header[0]);
    free_ogg_packet (vorbis->header[1]);
    free_ogg_packet (vorbis->header[2]);
    memset (vorbis->header, 0, sizeof (vorbis->header));
    vorbis->header [0] = copy_ogg_packet (&packet);
    ogg_stream_init (&vorbis->new_os, rand());

    codec->process_page = process_vorbis_page;
    codec->process = process_vorbis;
    plugin->set_tag = vorbis_set_tag;

    vorbis->bos_page.header = malloc (page->header_len + page->body_len);

    memcpy (vorbis->bos_page.header, page->header, page->header_len);
    vorbis->bos_page.header_len = page->header_len;

    vorbis->bos_page.body = vorbis->bos_page.header + page->header_len;
    memcpy (vorbis->bos_page.body, page->body, page->body_len);
    vorbis->bos_page.body_len = page->body_len;

    return codec;
}

/* The auth thread main loop. */
static void *auth_run_thread (void *arg)
{
    auth_t *auth = arg;

    INFO0 ("Authentication thread started");
    while (auth->running)
    {
        /* usually no clients are waiting, so don't bother taking locks */
        if (auth->head)
        {
            auth_client *auth_user;

            /* may become NULL before lock taken */
            thread_mutex_lock (&auth->lock);
            auth_user = (auth_client*)auth->head;
            if (auth_user == NULL)
            {
                thread_mutex_unlock (&auth->lock);
                continue;
            }
            DEBUG2 ("%d client(s) pending on %s", auth->pending_count, auth->mount);
            auth->head = auth_user->next;
            if (auth->head == NULL)
                auth->tailp = &auth->head;
            auth->pending_count--;
            thread_mutex_unlock (&auth->lock);
            auth_user->next = NULL;

            if (auth_user->process)
                auth_user->process (auth, auth_user);
            else
                ERROR0 ("client auth process not set");

            auth_client_free (auth_user);

            continue;
        }
        thread_sleep (150000);
    }
    INFO0 ("Authenication thread shutting down");
    return NULL;
}

static void *yp_update_thread(void *arg)
{
    if (!kitsune_is_updating()) { /**DSU control */
        INFO0("YP update thread started");
        yp_running = 1;
    }

    while (yp_running)
    {
      kitsune_update("yp_update"); /**DSU updatepoint */
        struct yp_server *server;

        thread_sleep (200000);

        /* do the YP communication */
        thread_rwlock_rlock (&yp_lock);
        server = (struct yp_server *)active_yps;
        while (server)
        {
            /* DEBUG1 ("trying %s", server->url); */
            yp_process_server (server);
            server = server->next;
        }
        thread_rwlock_unlock (&yp_lock);

        /* update the local YP structure */
        if (yp_update)
        {
            thread_rwlock_wlock (&yp_lock);
            check_servers ();
            server = (struct yp_server *)active_yps;
            while (server)
            {
                /* DEBUG1 ("Checking yps %s", server->url); */
                add_pending_yp (server);
                delete_marked_yp (server);
                server = server->next;
            }
            yp_update = 0;
            thread_rwlock_unlock (&yp_lock);
        }
    }
    thread_rwlock_destroy (&yp_lock);
    thread_mutex_destroy (&yp_pending_lock);
    /* free server and ypdata left */
    while (active_yps)
    {
        struct yp_server *server = (struct yp_server *)active_yps;
        active_yps = server->next;
        destroy_yp_server (server);
    }

    return NULL;
}