struct poolworker_state *poolworker_create(int numThreads) {
    struct poolworker_state *pw;
    int i;

    pw = safe_calloc(1, sizeof *pw);
    pw->p = safe_calloc(numThreads, sizeof *pw->p);
    pw->jobs_req = NULL;
    pw->jobs_active = NULL;

    pthread_mutex_init(&pw->lock, NULL);
    pthread_cond_init(&pw->cond_req, NULL);
    pthread_cond_init(&pw->cond_cmp, NULL);

    atomic_write(&pw->numRunning, 0);
    atomic_write(&pw->exit, 0);

    Log(LGPFX " creating %u threads\n", numThreads);

    for (i = 0; i < numThreads; i++) {
        pthread_create(&pw->p[i].tid, NULL, poolworker_main, pw);

        while (atomic_read(&pw->numRunning) != i + 1) {
            sched_yield();
        }
    }
    return pw;
}

int *get_adrcmd_from_key(int cxt, struct key_entry *key, int create) {
    flrn_assoc_key_cmd *pc;
    unsigned char hash;
    int i;
    if (key==NULL) return NULL;
    if (key->entry==ENTRY_ERROR_KEY) return NULL;
    hash=get_hash_key(key);
    pc=Flcmd_rev[hash];
    if (pc==NULL) {
        if (create) 
	    pc=Flcmd_rev[hash]=safe_calloc(1,sizeof(flrn_assoc_key_cmd));
	else return NULL;
    } else { 
	while (pc->next) {
            if (key_are_equal(&(pc->key),key)) break;
	    pc=pc->next;
	}
	if (key_are_equal(&(pc->key),key)) return &(pc->cmd[cxt]);
	if (create==0) return NULL;
	pc->next=safe_calloc(1,sizeof(flrn_assoc_key_cmd));
	pc=pc->next;
    }
    /* create = 1 */
    for (i=0;i<NUMBER_OF_CONTEXTS;i++) pc->cmd[i]=FLCMD_UNDEF;
    memcpy(&(pc->key),key,sizeof(struct key_entry));
    if (key->entry==ENTRY_ALLOCATED_STRING) 
	pc->key.value.allocstr=safe_flstrdup(key->value.allocstr);
    return &(pc->cmd[cxt]);
}

/*  
 * -- init_map
 *
 * initializes global shared map.
 *
 */
void
init_map(struct _como * m)
{
    int i; 

    memset(m, 0, sizeof(struct _como));
    m->whoami = SUPERVISOR;
    m->running = NORMAL; 
    m->logflags = DEFAULT_LOGFLAGS;
    m->mem_size = DEFAULT_MEMORY;
    m->maxfilesize = DEFAULT_FILESIZE;
    m->module_max = DEFAULT_MODULE_MAX;
    m->module_last = -1; 
    m->modules = safe_calloc(m->module_max, sizeof(module_t));
    for (i = 0; i < m->module_max; i++) 
	m->modules[i].status = MDL_UNUSED; 
    m->workdir = mkdtemp(strdup("/tmp/comoXXXXXX"));
    m->basedir = strdup(DEFAULT_BASEDIR);
    m->libdir = strdup(DEFAULT_LIBDIR);
    m->node = safe_calloc(1, sizeof(node_t)); 
    m->node->id = 0;
    m->node->name = strdup("CoMo Node");
    m->node->location = strdup("Unknown");
    m->node->type = strdup("Unknown");
    m->node->query_port = DEFAULT_QUERY_PORT;
    m->node_count = 1;
}

/*
 * Initialize a term, using a window of the given size.
 * Also prepare the "input queue" for "k" keypresses
 * By default, the cursor starts out "invisible"
 * By default, we "erase" using "black spaces"
 */
errr term_init(term *t, int w, int h, int k)
{
	int y;

	/* Wipe it */
	memset(t, 0, sizeof(term));


	/* Prepare the input queue */
	t->key_head = t->key_tail = 0;

	/* Determine the input queue size */
	t->key_size = k;

	/* Allocate the input queue */
	t->key_queue = safe_calloc(t->key_size, sizeof(char));


	/* Save the size */
	t->wid = w;
	t->hgt = h;

	/* Allocate change arrays */
	t->x1 = safe_calloc(h, sizeof(byte));
	t->x2 = safe_calloc(h, sizeof(byte));


	/* Allocate "displayed" */
	t->old = safe_calloc(1, sizeof(struct term_win));

	/* Initialize "displayed" */
	term_win_init(t->old, w, h);


	/* Allocate "requested" */
	t->scr = safe_calloc(1, sizeof(struct term_win));

	/* Initialize "requested" */
	term_win_init(t->scr, w, h);


	/* Assume change */
	for (y = 0; y < h; y++)
	{
		/* Assume change */
		t->x1[y] = 0;
		t->x2[y] = w - 1;
	}

	/* Assume change */
	t->y1 = 0;
	t->y2 = h - 1;

	/* Force "total erase" */
	t->total_erase = TRUE;


	/* Success */
	return (0);
}

/**
 * For a given client, calculate the master key and do key expansion
 * to determine the symmetric cypher key and IV salt, and hash key
 */
void calculate_client_keys(struct pr_group_list_t *group, int hostidx)
{
    unsigned char *seed, *prf_buf;
    int explen, len, seedlen;
    struct pr_destinfo_t *dest;

    dest = &group->destinfo[hostidx];

    explen = group->keylen + group->ivlen +
             group->hmaclen;
    seedlen = sizeof(group->rand1) * 2;
    seed = safe_calloc(seedlen, 1);
    prf_buf = safe_calloc(MASTER_LEN + explen + group->hmaclen, 1);

    memcpy(seed, group->rand1, sizeof(group->rand1));
    memcpy(seed + sizeof(group->rand1), dest->rand2,
            sizeof(dest->rand2));
    PRF(group->hashtype, MASTER_LEN, dest->premaster, dest->premaster_len,
            "master secret", seed, seedlen, prf_buf, &len);
    memcpy(dest->master,prf_buf, sizeof(dest->master));

    PRF(group->hashtype, explen, dest->master, sizeof(dest->master),
            "key expansion", seed, seedlen, prf_buf, &len);
    memcpy(dest->hmackey, prf_buf, group->hmaclen);
    memcpy(dest->key, prf_buf + group->hmaclen, group->keylen);
    memcpy(dest->salt, prf_buf + group->hmaclen + group->keylen, group->ivlen);

    free(seed);
    free(prf_buf);
}

/*  
 * -- init_map
 *
 * initializes global shared map.
 *
 */
void
init_map(struct _como * m)
{
    int i; 

    memset(m, 0, sizeof(struct _como));
    m->whoami = SUPERVISOR;
    m->runmode = RUNMODE_NORMAL; 
    m->logflags = DEFAULT_LOGFLAGS;
    m->mem_size = DEFAULT_MEMORY;
    m->maxfilesize = DEFAULT_FILESIZE;
    m->module_max = DEFAULT_MODULE_MAX;
    m->module_last = -1; 
    m->modules = safe_calloc(m->module_max, sizeof(module_t));
    for (i = 0; i < m->module_max; i++) 
	m->modules[i].status = MDL_UNUSED; 
    m->workdir = mkdtemp(strdup("/tmp/comoXXXXXX"));
    m->dbdir = strdup(DEFAULT_DBDIR);
    m->libdir = strdup(DEFAULT_LIBDIR);
    m->node = safe_calloc(1, sizeof(node_t)); 
    m->node[0].name = strdup("CoMo Node");
    m->node[0].location = strdup("Unknown");
    m->node[0].type = strdup("Unknown");
    m->node[0].query_port = DEFAULT_QUERY_PORT;
    m->node_count = 1;
    m->debug_sleep = 20;
    m->asnfile = NULL;
    m->live_thresh = TIME2TS(0, 10000); /* default 10 ms */
}

/**
 * Sends a KEYINFO_ACK in response to a KEYINFO
 */
void send_keyinfo_ack(struct group_list_t *group)
{
    unsigned char *buf, *encrypted;
    struct uftp_h *header;
    struct keyinfoack_h *keyinfo_ack;
    unsigned char *verifydata, *verify_hash, *verify_val;
    unsigned int payloadlen, hashlen;
    int verifylen, enclen, len;

    buf = safe_calloc(MAXMTU, 1);

    header = (struct uftp_h *)buf;
    keyinfo_ack = (struct keyinfoack_h *)(buf + sizeof(struct uftp_h));

    set_uftp_header(header, KEYINFO_ACK, group);
    keyinfo_ack->func = KEYINFO_ACK;
    keyinfo_ack->hlen = sizeof(struct keyinfoack_h) / 4;

    verifydata = build_verify_data(group, &verifylen);
    if (!verifydata) {
        glog0(group, "Error getting verify data");
        send_abort(group, "Error getting verify data");
        free(buf);
        return;
    }

    verify_hash = safe_calloc(group->hmaclen, 1);
    verify_val = safe_calloc(VERIFY_LEN + group->hmaclen, 1);
    hash(group->hashtype, verifydata, verifylen, verify_hash, &hashlen);
    PRF(group->hashtype, VERIFY_LEN, group->groupmaster,
            sizeof(group->groupmaster), "client finished",
            verify_hash, hashlen, verify_val, &len);
    memcpy(keyinfo_ack->verify_data, verify_val, VERIFY_LEN);
    free(verifydata);
    free(verify_hash);
    free(verify_val);

    payloadlen = sizeof(struct keyinfoack_h);
    encrypted = NULL;
    if (!encrypt_and_sign(buf, &encrypted, payloadlen, &enclen, group->keytype,
            group->groupkey, group->groupsalt, &group->ivctr, group->ivlen,
            group->hashtype, group->grouphmackey, group->hmaclen,group->sigtype,
            group->keyextype, group->client_privkey,group->client_privkeylen)) {
        glog0(group, "Error encrypting KEYINFO_ACK");
        free(buf);
        return;
    }
    payloadlen = enclen + sizeof(struct uftp_h);

    if (nb_sendto(listener, encrypted, payloadlen, 0,
               (struct sockaddr *)&(group->replyaddr),
               family_len(group->replyaddr)) == SOCKET_ERROR) {
        gsockerror(group, "Error sending KEYINFO_ACK");
    } else {
        glog2(group, "KEYINFO_ACK sent");
    }
    free(encrypted);
    free(buf);
}

// crate a stack of <no_element> size
nodes_stack * Create_Stack(int no_elements) {
  nodes_stack * s = (nodes_stack *)safe_calloc(1, sizeof(nodes_stack));
  assert(no_elements);
  s->top = 0;
  s->max_size = no_elements;
  s->nodes = (int *)safe_calloc(no_elements, sizeof(int));
  return s;
}

static QUERY *run_query (char *s, int quiet)
{
        FILE *fp;
        QUERY *first = NULL;
        QUERY *cur = NULL;
        char cmd[_POSIX_PATH_MAX];
        char *buf = NULL;
        size_t buflen;
        int dummy = 0;
        char msg[STRING];
        char *p;
        pid_t thepid;

        mutt_expand_file_fmt (cmd, sizeof(cmd), QueryCmd, s);

        if ((thepid = mutt_create_filter (cmd, NULL, &fp, NULL)) < 0) {
                dprint (1, (debugfile, "unable to fork command: %s", cmd));
                return 0;
        }
        if (!quiet)
                mutt_message _("Waiting for response...");
        fgets (msg, sizeof (msg), fp);
        if ((p = strrchr (msg, '\n')))
                *p = '\0';
        while ((buf = mutt_read_line (buf, &buflen, fp, &dummy, 0)) != NULL) {
                if ((p = strtok(buf, "\t\n"))) {
                        if (first == NULL) {
                                first = (QUERY *) safe_calloc (1, sizeof (QUERY));
                                cur = first;
                        }
                        else {
                                cur->next = (QUERY *) safe_calloc (1, sizeof (QUERY));
                                cur = cur->next;
                        }

                        cur->addr = rfc822_parse_adrlist (cur->addr, p);
                        p = strtok(NULL, "\t\n");
                        if (p) {
                                cur->name = safe_strdup (p);
                                p = strtok(NULL, "\t\n");
                                if (p)
                                        cur->other = safe_strdup (p);
                        }
                }
        }
        FREE (&buf);
        safe_fclose (&fp);
        if (mutt_wait_filter (thepid)) {
                dprint (1, (debugfile, "Error: %s\n", msg));
                if (!quiet)  mutt_error ("%s", msg);
        }
        else {
                if (!quiet)
                        mutt_message ("%s", msg);
        }

        return first;
}

/**
 * Handles an incoming KEYINFO_ACK message from a client
 */
void handle_keyinfo_ack(struct pr_group_list_t *group, int hostidx,
                        const unsigned char *message, unsigned meslen)
{
    const struct keyinfoack_h *keyinfoack;
    unsigned char *verifydata, *verify_hash, *verify_test;
    int verifylen, len, dupmsg;
    unsigned int hashlen;
    struct pr_destinfo_t *dest;

    keyinfoack = (const struct keyinfoack_h *)message;
    dest = &group->destinfo[hostidx];

    if ((meslen < (keyinfoack->hlen * 4U)) ||
            ((keyinfoack->hlen * 4U) < sizeof(struct keyinfoack_h))) {
        glog1(group, "Rejecting KEYINFO_ACK from %s: invalid message size",
                     dest->name);
        send_downstream_abort(group, dest->id, "Invalid message size", 0);
        return;
    }

    if (!(verifydata = build_verify_data(group, hostidx, &verifylen,1))) {
        glog1(group, "Rejecting KEYINFO_ACK from %s: "
                     "error exporting client public key", dest->name);
        return;
    }
    verify_hash = safe_calloc(group->hmaclen, 1);
    verify_test = safe_calloc(VERIFY_LEN + group->hmaclen, 1);
    hash(group->hashtype, verifydata, verifylen, verify_hash, &hashlen);
    PRF(group->hashtype, VERIFY_LEN, group->groupmaster,
            sizeof(group->groupmaster), "client finished",
            verify_hash, hashlen, verify_test, &len);
    if (memcmp(keyinfoack->verify_data, verify_test, VERIFY_LEN)) {
        glog1(group, "Rejecting KEYINFO_ACK from %s: verify data mismatch",
                     dest->name);
        free(verifydata);
        free(verify_hash);
        free(verify_test);
        return;
    }

    free(verifydata);
    free(verify_hash);
    free(verify_test);

    dupmsg = (dest->state == PR_CLIENT_READY);
    glog2(group, "Received KEYINFO_ACK%s from %s", dupmsg ? "+" : "",
                 dest->name);
    dest->state = PR_CLIENT_READY;
    if (!check_unfinished_clients(group, 0)) {
        group->phase = PR_PHASE_RECEIVING;
    }
}

struct bitfield *
make_bitfield(const UChar *word, const UChar *alphabet)
{
    int alphabetlength = u_strlen(alphabet);
    int *freqs = safe_calloc(alphabetlength, sizeof(int));
    struct bitfield *bf;
    int maxfreq = 0;
    int index;

    if (freqs == NULL)
        return NULL;

    while (*word) {
        const UChar *p = alphabet;
        for (index = 0; *p && (*p != *word); p++, index++) {
        }
        if (*p) {
            freqs[index]++;
            if (freqs[index] > maxfreq) {
                maxfreq = freqs[index];
            }
        } else {
            /* character not in alphabet */
            free(freqs);
            return NULL;
        }
        word++;
    }

    bf = safe_calloc(1, sizeof(struct bitfield));
    if (bf != NULL) {
        bitpattern thisbit;

        bf->depth = maxfreq;
        bf->bits = safe_calloc(sizeof(bitpattern), maxfreq);
        thisbit = 1;

        for (index = 0; index <alphabetlength; index++) {
            bitpattern *bits = bf->bits;
            while (freqs[index] > 0) {
                *bits |= thisbit;
                bits++;
                freqs[index]--;
            }
            thisbit <<= 1;        
        }
    }
    free(freqs);
    return bf;
}

static int ssl_socket_open (CONNECTION * conn)
{
  sslsockdata *data;
  int maxbits;

  if (raw_socket_open (conn) < 0)
    return -1;

  data = (sslsockdata *) safe_calloc (1, sizeof (sslsockdata));
  conn->sockdata = data;

  data->ctx = SSL_CTX_new (SSLv23_client_method ());

  /* disable SSL protocols as needed */
  if (!option(OPTTLSV1))
  {
    SSL_CTX_set_options(data->ctx, SSL_OP_NO_TLSv1);
  }
  /* TLSv1.1/1.2 support was added in OpenSSL 1.0.1, but some OS distros such
   * as Fedora 17 are on OpenSSL 1.0.0.
   */
#ifdef SSL_OP_NO_TLSv1_1
  if (!option(OPTTLSV1_1))
  {
    SSL_CTX_set_options(data->ctx, SSL_OP_NO_TLSv1_1);
  }
#endif
#ifdef SSL_OP_NO_TLSv1_2
  if (!option(OPTTLSV1_2))
  {
    SSL_CTX_set_options(data->ctx, SSL_OP_NO_TLSv1_2);
  }
#endif
  if (!option(OPTSSLV2))
  {
    SSL_CTX_set_options(data->ctx, SSL_OP_NO_SSLv2);
  }
  if (!option(OPTSSLV3))
  {
    SSL_CTX_set_options(data->ctx, SSL_OP_NO_SSLv3);
  }

  ssl_get_client_cert(data, conn);

  data->ssl = SSL_new (data->ctx);
  SSL_set_fd (data->ssl, conn->fd);

  if (ssl_negotiate(conn, data))
  {
    mutt_socket_close (conn);
    return -1;
  }

  data->isopen = 1;

  conn->ssf = SSL_CIPHER_get_bits (SSL_get_current_cipher (data->ssl),
    &maxbits);

  return 0;
}

/**
 * Encrypts a small block of data with an RSA public key.
 * Output buffer must be at least the key size.
 */
int RSA_encrypt(RSA_key_t rsa, const unsigned char *from, unsigned int fromlen,
                unsigned char *to, unsigned int *tolen)
{
    DWORD _tolen;
    int flags;
    unsigned int i;
    unsigned char *outbuf;

    if (RSA_keylen(rsa) * 8 < 768) {
        flags = 0;
    } else {
        flags = CRYPT_OAEP;
    }

    outbuf = safe_calloc(RSA_keylen(rsa), 1);
    memcpy(outbuf, from, fromlen);
    _tolen = fromlen;
    if (!CryptEncrypt(rsa, 0, 1, flags, outbuf, &_tolen, RSA_keylen(rsa))) {
        mserror("CryptEncrypt failed");
        free(outbuf);
        return 0;
    }
    *tolen = _tolen;

    // CryptoAPI returns ciphertext in little endian, so reverse the bytes
    for (i = 0; i < _tolen; i++) {
        to[i] = outbuf[_tolen - i - 1];
    }

    free(outbuf);
    return 1;
}

/*
 * -- sniffer_init
 * 
 */
static sniffer_t *
sniffer_init(const char * device, const char * args)
{
    struct bpf_me *me;
    
    me = safe_calloc(1, sizeof(struct bpf_me));

    me->sniff.max_pkts = 8192;
    me->sniff.flags = SNIFF_SELECT | SNIFF_SHBUF;
    me->device = device;
    
    /* create the capture buffer */
    if (capbuf_init(&me->capbuf, args, NULL, BPF_MIN_BUFSIZE,
		    BPF_MAX_BUFSIZE) < 0)
	goto error;

    /* create the pkt_t buffer */
    if (capbuf_init(&me->pktbuf, args, "pktbuf=", BPF_MIN_PKTBUFSIZE,
		    BPF_MAX_PKTBUFSIZE) < 0)
	goto error;

    me->read_size = me->capbuf.size / 2;
    me->min_proc_size = BPF_DEFAULT_MIN_PROC_SIZE;
    
    return (sniffer_t *) me;
error:
    free(me);
    return NULL; 
}

/*
 * Restore the "requested" contents (see above).
 *
 * Every "Term_save()" should match exactly one "Term_load()"
 */
errr Term_load(void)
{
	int y;

	int w = Term->wid;
	int h = Term->hgt;

	/* Create */
	if (!Term->mem)
	{
		/* Allocate window */
		Term->mem = safe_calloc(1, sizeof(struct term_win));

		/* Initialize window */
		term_win_init(Term->mem, w, h);
	}

	/* Load */
	term_win_copy(Term->scr, Term->mem, w, h);

	/* Assume change */
	for (y = 0; y < h; y++)
	{
		/* Assume change */
		Term->x1[y] = 0;
		Term->x2[y] = w - 1;
	}

	/* Assume change */
	Term->y1 = 0;
	Term->y2 = h - 1;

	/* Success */
	return (0);
}

static void SCEDA_hashmap_resize(SCEDA_HashMap *hmap, int buckets) {
  if((buckets <= 1) || (buckets == hmap->buckets)) {
    return;
  }

  SCEDA_ListMap (*new_table)[] = safe_calloc(buckets, sizeof(SCEDA_ListMap));
  int i;
  for(i = 0; i < buckets; i++) {
    SCEDA_listmap_init(nth_map(new_table, i), hmap->delete_key,
		 hmap->delete_value, hmap->match_key);
  }

  for(i = 0; i < hmap->buckets; i++) {
    SCEDA_ListMap *map = SCEDA_hashmap_nth_map(hmap, i);
    while(SCEDA_listmap_size(map) > 0) {
      void *key;
      void *value;
      SCEDA_listmap_remove_head(map, &key, &value);
      int j = new_hash_code(hmap, key, buckets);
      SCEDA_listmap_add_tail(nth_map(new_table, j), key, value);
    }
    SCEDA_listmap_cleanup(map);
  }

  free(hmap->table);
  hmap->table = new_table;
  hmap->buckets = buckets;
}

bool key_sign(struct key *k, const void *data, size_t datalen, uint8 **sig,
              size_t *siglen)

{
  unsigned int len;
  uint8 *sig0;
  int res;

  ASSERT(sig);
  ASSERT(siglen);

  len = ECDSA_size(k->key);
  sig0 = safe_calloc(1, len);

  res = ECDSA_sign(0, data, datalen, sig0, &len, k->key);
  if (res != 1) {
    NOT_TESTED();
    free(sig0);
    return 0;
  }
  *sig = sig0;
  *siglen = len;

  return 1;
}

int deserialize_str_alloc(struct buff *buf, char **str, size_t *len) {
  uint64 length;
  int res;

  *str = NULL;
  if (len) {
    *len = 0;
  }

  res = deserialize_varint(buf, &length);
  if (res) {
    return res;
  }
  if (length == 0) {
    return 0;
  }

  *str = safe_calloc(1, length + 1);
  if (len) {
    *len = length;
  }

  if (deserialize_bytes(buf, *str, length)) {
    free(*str);
    *str = NULL;
    if (len) {
      *len = 0;
    }
    return 1;
  }
  return 0;
}

static void add_to_list (LIST **list, const char *str)
{
  LIST *t, *last = NULL;

  /* check to make sure the item is not already on this list */
  for (last = *list; last; last = last->next)
  {
    if (mutt_strcasecmp (str, last->data) == 0)
    {
      /* already on the list, so just ignore it */
      last = NULL;
      break;
    }
    if (!last->next)
      break;
  }

  if (!*list || last)
  {
    t = (LIST *) safe_calloc (1, sizeof (LIST));
    t->data = safe_strdup (str);
    if (last)
    {
      last->next = t;
      last = last->next;
    }
    else
      *list = last = t;
  }
}

smap *smap_new() {
    smap *ret = safe_malloc(sizeof(smap));
    ret->num_buckets = INIT_CAPACITY;
    ret->num_pairs = 0;
    ret->buckets = safe_calloc(ret->num_buckets * sizeof(bucket));
    return ret;
}