Function dohsql_distribute

db/dohsql.c:1162–1251

}

int dohsql_distribute(dohsql_node_t *node)
{
    GET_CLNT;
    dohsql_t *conns;
    int i, rc;
    int clnt_nparams;
    int flags = 0;

    if (gbl_dohsql_max_threads && node->nnodes > gbl_dohsql_max_threads) {
        return SHARD_ERR_TOOMANYTHR;
    }

    clnt_nparams = param_count(clnt);
    if (clnt_nparams != node->nparams) {
        return SHARD_ERR_PARAMS;
    }

    rc = _add_parallel_load(clnt);
    if (rc)
        return rc;

    /* setup communication queue */
    conns = (dohsql_t *)calloc(
        1, sizeof(dohsql_t) + node->nnodes * sizeof(dohsql_connector_t));
    if (!conns)
        return SHARD_ERR_MALLOC;
    conns->conns = (dohsql_connector_t *)(conns + 1);
    conns->nconns = node->nnodes;
    conns->ncols = node->ncols;
    conns->nparams = node->nparams;

    if (node->order_size) {
        if (order_init(conns, node)) {
            free(conns);
            return SHARD_ERR_MALLOC;
        }
        flags = THDPOOL_FORCE_DISPATCH;
    }
    /* there is a slack to allow non-coordinator tasks to drain;
     * it is still possible to fill the sql queue; force the 
     * worker shards on the queue in any case
     */
    flags |= THDPOOL_FORCE_QUEUE;
    clnt->conns = conns;
    /* augment interface */
    _master_clnt_set(clnt);

    /* start peers */
    for (i = 0; i < conns->nconns; i++) {
        struct param_data *params;
        int nparams;
        _save_params(node->nodes[i], &params, &nparams);
        if ((rc = _shard_connect(clnt, &conns->conns[i], node->nodes[i]->sql,
                                 nparams, params)) != 0)
            return rc;

        if (i > 0) {
            struct string_ref *sr = create_string_ref(node->nodes[i]->sql);