/* Gearman server and library * Copyright (C) 2008 Brian Aker, Eric Day * All rights reserved. * * Use and distribution licensed under the BSD license. See * the COPYING file in the parent directory for full text. */ /** * @file * @brief Server Definitions */ #include #include #include #include #include #include /* * Private declarations */ #define TEXT_SUCCESS "OK\r\n" #define TEXT_ERROR_ARGS "ERR INVALID_ARGUMENTS An+incomplete+set+of+arguments+was+sent+to+this+command+%.*s\r\n" #define TEXT_ERROR_CREATE_FUNCTION "ERR CREATE_FUNCTION %.*s\r\n" #define TEXT_ERROR_UNKNOWN_COMMAND "ERR UNKNOWN_COMMAND Unknown+server+command%.*s\r\n" #define TEXT_ERROR_INTERNAL_ERROR "ERR UNKNOWN_ERROR\r\n" /** * @addtogroup gearman_server_private Private Server Functions * @ingroup gearman_server * @{ */ /** * Add job to queue wihle replaying queue during startup. */ static gearmand_error_t _queue_replay_add(gearman_server_st *server, void *context, const char *unique, size_t unique_size, const char *function_name, size_t function_name_size, const void *data, size_t data_size, gearmand_job_priority_t priority, int64_t when); /** * Queue an error packet. */ static gearmand_error_t _server_error_packet(gearman_server_con_st *server_con, const char *error_code, const char *error_string); /** * Process text commands for a connection. */ static gearmand_error_t _server_run_text(gearman_server_con_st *server_con, gearmand_packet_st *packet); /** * Send work result packets with data back to clients. */ static gearmand_error_t _server_queue_work_data(gearman_server_job_st *server_job, gearmand_packet_st *packet, gearman_command_t command); /** @} */ /* * Public definitions */ gearmand_error_t gearman_server_run_command(gearman_server_con_st *server_con, gearmand_packet_st *packet) { gearmand_error_t ret; gearman_server_job_st *server_job; char job_handle[GEARMAND_JOB_HANDLE_SIZE]; char option[GEARMAN_OPTION_SIZE]; gearman_server_client_st *server_client= NULL; char numerator_buffer[11]; /* Max string size to hold a uint32_t. */ char denominator_buffer[11]; /* Max string size to hold a uint32_t. */ gearmand_job_priority_t priority; int checked_length; if (packet->magic == GEARMAN_MAGIC_RESPONSE) { return _server_error_packet(server_con, "bad_magic", "Request magic expected"); } gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "%15s:%5s packet command %s", server_con->con.context == NULL ? "-" : server_con->con.context->host, server_con->con.context == NULL ? "-" : server_con->con.context->port, gearmand_strcommand(packet)); switch (packet->command) { /* Client/worker requests. */ case GEARMAN_COMMAND_ECHO_REQ: /* Reuse the data buffer and just shove the data back. */ ret= gearman_server_io_packet_add(server_con, true, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_ECHO_RES, packet->data, packet->data_size, NULL); if (gearmand_failed(ret)) { return gearmand_gerror("gearman_server_io_packet_add", ret); } packet->options.free_data= false; break; case GEARMAN_COMMAND_SUBMIT_REDUCE_JOB: // Reduce request server_client= gearman_server_client_add(server_con); if (server_client == NULL) { return GEARMAN_MEMORY_ALLOCATION_FAILURE; } case GEARMAN_COMMAND_SUBMIT_REDUCE_JOB_BACKGROUND: { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Received reduce submission, Partitioner: %.*s(%lu) Reducer: %.*s(%lu) Unique: %.*s(%lu) with %d arguments", packet->arg_size[0] -1, packet->arg[0], packet->arg_size[0] -1, packet->arg_size[2] -1, packet->arg[2], packet->arg_size[2] -1, // reducer packet->arg_size[1] -1, packet->arg[1], packet->arg_size[1] -1, (int)packet->argc); if (packet->arg_size[2] -1 > GEARMAN_UNIQUE_SIZE) { gearman_server_client_free(server_client); gearmand_gerror("unique value too large", GEARMAN_ARGUMENT_TOO_LARGE); return _server_error_packet(server_con, "job failure", "Unique value too large"); } gearmand_job_priority_t map_priority= GEARMAND_JOB_PRIORITY_NORMAL; /* Schedule job. */ server_job= gearman_server_job_add_reducer(Server, (char *)(packet->arg[0]), packet->arg_size[0] -1, // Function (char *)(packet->arg[1]), packet->arg_size[1] -1, // unique (char *)(packet->arg[2]), packet->arg_size[2] -1, // reducer packet->data, packet->data_size, map_priority, server_client, &ret, 0); if (gearmand_success(ret)) { packet->options.free_data= false; } else if (ret == GEARMAN_JOB_QUEUE_FULL) { gearman_server_client_free(server_client); return _server_error_packet(server_con, "queue_full", "Job queue is full"); } else if (ret != GEARMAN_JOB_EXISTS) { gearman_server_client_free(server_client); gearmand_gerror("gearman_server_job_add", ret); return ret; } /* Queue the job created packet. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_CREATED, server_job->job_handle, (size_t)strlen(server_job->job_handle), NULL); if (gearmand_failed(ret)) { gearman_server_client_free(server_client); gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } gearmand_log_notice(GEARMAN_DEFAULT_LOG_PARAM,"accepted,%.*s,%.*s,%.*s", packet->arg_size[0] -1, packet->arg[0], // Function packet->arg_size[1] -1, packet->arg[1], // unique packet->arg_size[2] -1, packet->arg[2]); // reducer } break; /* Client requests. */ case GEARMAN_COMMAND_SUBMIT_JOB: case GEARMAN_COMMAND_SUBMIT_JOB_BG: case GEARMAN_COMMAND_SUBMIT_JOB_HIGH: case GEARMAN_COMMAND_SUBMIT_JOB_HIGH_BG: case GEARMAN_COMMAND_SUBMIT_JOB_LOW: case GEARMAN_COMMAND_SUBMIT_JOB_LOW_BG: case GEARMAN_COMMAND_SUBMIT_JOB_EPOCH: if (packet->command == GEARMAN_COMMAND_SUBMIT_JOB || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_BG || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_EPOCH) { priority= GEARMAND_JOB_PRIORITY_NORMAL; } else if (packet->command == GEARMAN_COMMAND_SUBMIT_JOB_HIGH || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_HIGH_BG) { priority= GEARMAND_JOB_PRIORITY_HIGH; } else { priority= GEARMAND_JOB_PRIORITY_LOW; } if (packet->command == GEARMAN_COMMAND_SUBMIT_JOB_BG || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_HIGH_BG || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_LOW_BG || packet->command == GEARMAN_COMMAND_SUBMIT_JOB_EPOCH) { server_client= NULL; } else { server_client= gearman_server_client_add(server_con); if (server_client == NULL) { return GEARMAN_MEMORY_ALLOCATION_FAILURE; } } gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Received submission, %.*s/%.*s with %d arguments", packet->arg_size[0], packet->arg[0], packet->arg_size[1], packet->arg[1], (int)packet->argc); int64_t when= 0; if (packet->command == GEARMAN_COMMAND_SUBMIT_JOB_EPOCH) { sscanf((char *)packet->arg[2], "%lld", (long long *)&when); gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Received EPOCH job submission, %.*s/%.*s, with data for %jd at %jd, args %d", packet->arg_size[0], packet->arg[0], packet->arg_size[1], packet->arg[1], when, time(NULL), (int)packet->argc); } if (packet->arg_size[1] -1 > GEARMAN_UNIQUE_SIZE) { gearmand_gerror("unique value too large", GEARMAN_ARGUMENT_TOO_LARGE); gearman_server_client_free(server_client); return _server_error_packet(server_con, "job failure", "Unique value too large"); } /* Schedule job. */ server_job= gearman_server_job_add(Server, (char *)(packet->arg[0]), packet->arg_size[0] -1, // Function (char *)(packet->arg[1]), packet->arg_size[1] -1, // unique packet->data, packet->data_size, priority, server_client, &ret, when); if (gearmand_success(ret)) { packet->options.free_data= false; } else if (ret == GEARMAN_JOB_QUEUE_FULL) { gearman_server_client_free(server_client); return _server_error_packet(server_con, "queue_full", "Job queue is full"); } else if (ret != GEARMAN_JOB_EXISTS) { gearman_server_client_free(server_client); gearmand_gerror("gearman_server_job_add", ret); return ret; } /* Queue the job created packet. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_CREATED, server_job->job_handle, (size_t)strlen(server_job->job_handle), NULL); if (gearmand_failed(ret)) { gearman_server_client_free(server_client); gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } gearmand_log_notice(GEARMAN_DEFAULT_LOG_PARAM,"accepted,%.*s,%.*s,%jd", packet->arg_size[0], packet->arg[0], // Function packet->arg_size[1], packet->arg[1], // Unique when); break; case GEARMAN_COMMAND_GET_STATUS: { /* This may not be NULL terminated, so copy to make sure it is. */ checked_length= snprintf(job_handle, GEARMAND_JOB_HANDLE_SIZE, "%.*s", (int)(packet->arg_size[0]), (char *)(packet->arg[0])); if (checked_length >= GEARMAND_JOB_HANDLE_SIZE || checked_length < 0) { gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "snprintf(%d)", checked_length); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } server_job= gearman_server_job_get(Server, job_handle, NULL); /* Queue status result packet. */ if (server_job == NULL) { ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_STATUS_RES, job_handle, (size_t)(strlen(job_handle) + 1), "0", (size_t)2, "0", (size_t)2, "0", (size_t)2, "0", (size_t)1, NULL); } else { checked_length= snprintf(numerator_buffer, sizeof(numerator_buffer), "%u", server_job->numerator); if ((size_t)checked_length >= sizeof(numerator_buffer) || checked_length < 0) { gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "snprintf(%d)", checked_length); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } checked_length= snprintf(denominator_buffer, sizeof(denominator_buffer), "%u", server_job->denominator); if ((size_t)checked_length >= sizeof(denominator_buffer) || checked_length < 0) { gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "snprintf(%d)", checked_length); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_STATUS_RES, job_handle, (size_t)(strlen(job_handle) + 1), "1", (size_t)2, server_job->worker == NULL ? "0" : "1", (size_t)2, numerator_buffer, (size_t)(strlen(numerator_buffer) + 1), denominator_buffer, (size_t)strlen(denominator_buffer), NULL); } if (ret != GEARMAN_SUCCESS) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } } break; case GEARMAN_COMMAND_OPTION_REQ: /* This may not be NULL terminated, so copy to make sure it is. */ checked_length= snprintf(option, GEARMAN_OPTION_SIZE, "%.*s", (int)(packet->arg_size[0]), (char *)(packet->arg[0])); if (checked_length >= GEARMAN_OPTION_SIZE || checked_length < 0) { gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "snprintf(%d)", checked_length); return _server_error_packet(server_con, "unknown_option", "Server does not recognize given option"); } if (strcasecmp(option, "exceptions") == 0) { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "'exceptions'"); server_con->is_exceptions= true; } else { return _server_error_packet(server_con, "unknown_option", "Server does not recognize given option"); } ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_OPTION_RES, packet->arg[0], packet->arg_size[0], NULL); if (ret != GEARMAN_SUCCESS) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } break; /* Worker requests. */ case GEARMAN_COMMAND_CAN_DO: if (gearman_server_worker_add(server_con, (char *)(packet->arg[0]), packet->arg_size[0], 0) == NULL) { return GEARMAN_MEMORY_ALLOCATION_FAILURE; } break; case GEARMAN_COMMAND_CAN_DO_TIMEOUT: if (gearman_server_worker_add(server_con, (char *)(packet->arg[0]), packet->arg_size[0] - 1, (in_port_t)atoi((char *)(packet->arg[1]))) == NULL) { return GEARMAN_MEMORY_ALLOCATION_FAILURE; } break; case GEARMAN_COMMAND_CANT_DO: gearman_server_con_free_worker(server_con, (char *)(packet->arg[0]), packet->arg_size[0]); break; case GEARMAN_COMMAND_RESET_ABILITIES: gearman_server_con_free_workers(server_con); break; case GEARMAN_COMMAND_PRE_SLEEP: server_job= gearman_server_job_peek(server_con); if (server_job == NULL) { server_con->is_sleeping= true; /* Remove any timeouts while sleeping */ gearman_server_con_delete_timeout(server_con); } else { /* If there are jobs that could be run, queue a NOOP packet to wake the worker up. This could be the result of a race codition. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_NOOP, NULL); if (ret != GEARMAN_SUCCESS) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } } break; case GEARMAN_COMMAND_GRAB_JOB: case GEARMAN_COMMAND_GRAB_JOB_UNIQ: case GEARMAN_COMMAND_GRAB_JOB_ALL: server_con->is_sleeping= false; server_con->is_noop_sent= false; server_job= gearman_server_job_take(server_con); if (server_job == NULL) { /* No jobs found, queue no job packet. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_NO_JOB, NULL); } else if (packet->command == GEARMAN_COMMAND_GRAB_JOB_UNIQ) { /* We found a runnable job, queue job assigned packet and take the job off the queue. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_ASSIGN_UNIQ, server_job->job_handle, (size_t)(strlen(server_job->job_handle) + 1), server_job->function->function_name, server_job->function->function_name_size + 1, server_job->unique, (size_t)(strlen(server_job->unique) + 1), server_job->data, server_job->data_size, NULL); } else if (packet->command == GEARMAN_COMMAND_GRAB_JOB_ALL and server_job->reducer) { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Sending reduce submission, Partitioner: %.*s(%lu) Reducer: %.*s(%lu) Unique: %.*s(%lu) with data sized (%lu)" , server_job->function->function_name_size, server_job->function->function_name, server_job->function->function_name_size, strlen(server_job->reducer), server_job->reducer, strlen(server_job->reducer), strlen(server_job->unique), server_job->unique, strlen(server_job->unique), (unsigned long)server_job->data_size); /* We found a runnable job, queue job assigned packet and take the job off the queue. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_ASSIGN_ALL, server_job->job_handle, (size_t)(strlen(server_job->job_handle) + 1), server_job->function->function_name, server_job->function->function_name_size + 1, server_job->unique, (size_t)(strlen(server_job->unique) + 1), server_job->reducer, (size_t)(strlen(server_job->reducer) +1), server_job->data, server_job->data_size, NULL); } else if (packet->command == GEARMAN_COMMAND_GRAB_JOB_ALL) { /* We found a runnable job, queue job assigned packet and take the job off the queue. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_ASSIGN_UNIQ, server_job->job_handle, (size_t)(strlen(server_job->job_handle) +1), server_job->function->function_name, server_job->function->function_name_size +1, server_job->unique, (size_t)(strlen(server_job->unique) +1), server_job->data, server_job->data_size, NULL); } else { /* Same, but without unique ID. */ ret= gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_JOB_ASSIGN, server_job->job_handle, (size_t)(strlen(server_job->job_handle) + 1), server_job->function->function_name, server_job->function->function_name_size + 1, server_job->data, server_job->data_size, NULL); } if (gearmand_failed(ret)) { gearmand_gerror("gearman_server_io_packet_add", ret); if (server_job) { return gearman_server_job_queue(server_job); } return ret; } /* Since job is assigned, we should respect function timeout */ if (server_job != NULL) { gearman_server_con_add_job_timeout(server_con, server_job); } break; case GEARMAN_COMMAND_WORK_DATA: case GEARMAN_COMMAND_WORK_WARNING: server_job= gearman_server_job_get(Server, (char *)(packet->arg[0]), server_con); if (server_job == NULL) { return _server_error_packet(server_con, "job_not_found", "Job given in work result not found"); } /* Queue the data/warning packet for all clients. */ ret= _server_queue_work_data(server_job, packet, packet->command); if (gearmand_failed(ret)) { gearmand_gerror("_server_queue_work_data", ret); return ret; } break; case GEARMAN_COMMAND_WORK_STATUS: server_job= gearman_server_job_get(Server, (char *)(packet->arg[0]), server_con); if (server_job == NULL) { return _server_error_packet(server_con, "job_not_found", "Job given in work result not found"); } /* Update job status. */ server_job->numerator= (uint32_t)atoi((char *)(packet->arg[1])); /* This may not be NULL terminated, so copy to make sure it is. */ checked_length= snprintf(denominator_buffer, sizeof(denominator_buffer), "%.*s", (int)(packet->arg_size[2]), (char *)(packet->arg[2])); if ((size_t)checked_length > sizeof(denominator_buffer) || checked_length < 0) { gearmand_error("snprintf"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } server_job->denominator= (uint32_t)atoi(denominator_buffer); /* Queue the status packet for all clients. */ for (server_client= server_job->client_list; server_client; server_client= server_client->job_next) { ret= gearman_server_io_packet_add(server_client->con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_WORK_STATUS, packet->arg[0], packet->arg_size[0], packet->arg[1], packet->arg_size[1], packet->arg[2], packet->arg_size[2], NULL); if (gearmand_failed(ret)) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } } break; case GEARMAN_COMMAND_WORK_COMPLETE: server_job= gearman_server_job_get(Server, (char *)(packet->arg[0]), server_con); if (server_job == NULL) { return _server_error_packet(server_con, "job_not_found", "Job given in work result not found"); } /* Queue the complete packet for all clients. */ ret= _server_queue_work_data(server_job, packet, GEARMAN_COMMAND_WORK_COMPLETE); if (gearmand_failed(ret)) { gearmand_gerror("_server_queue_work_data", ret); return ret; } /* Remove from persistent queue if one exists. */ if (server_job->job_queued && Server->queue._done_fn != NULL) { ret= (*(Server->queue._done_fn))(Server, (void *)Server->queue._context, server_job->unique, (size_t)strlen(server_job->unique), server_job->function->function_name, server_job->function->function_name_size); if (gearmand_failed(ret)) { gearmand_gerror("Remove from persistent queue", ret); return ret; } } /* Job is done, remove it. */ gearman_server_job_free(server_job); break; case GEARMAN_COMMAND_WORK_EXCEPTION: server_job= gearman_server_job_get(Server, (char *)(packet->arg[0]), server_con); if (server_job == NULL) { return _server_error_packet(server_con, "job_not_found", "Job given in work result not found"); } /* Queue the exception packet for all clients. */ ret= _server_queue_work_data(server_job, packet, GEARMAN_COMMAND_WORK_EXCEPTION); if (gearmand_failed(ret)) { gearmand_gerror("_server_queue_work_data", ret); return ret; } break; case GEARMAN_COMMAND_WORK_FAIL: /* This may not be NULL terminated, so copy to make sure it is. */ checked_length= snprintf(job_handle, GEARMAND_JOB_HANDLE_SIZE, "%.*s", (int)(packet->arg_size[0]), (char *)(packet->arg[0])); if (checked_length >= GEARMAND_JOB_HANDLE_SIZE || checked_length < 0) { return _server_error_packet(server_con, "job_name_too_large", "Error occured due to GEARMAND_JOB_HANDLE_SIZE being too small from snprintf"); } server_job= gearman_server_job_get(Server, job_handle, server_con); if (server_job == NULL) { return _server_error_packet(server_con, "job_not_found", "Job given in work result not found"); } /* Queue the fail packet for all clients. */ for (server_client= server_job->client_list; server_client; server_client= server_client->job_next) { ret= gearman_server_io_packet_add(server_client->con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_WORK_FAIL, packet->arg[0], packet->arg_size[0], NULL); if (gearmand_failed(ret)) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } } /* Remove from persistent queue if one exists. */ if (server_job->job_queued && Server->queue._done_fn != NULL) { ret= (*(Server->queue._done_fn))(Server, (void *)Server->queue._context, server_job->unique, (size_t)strlen(server_job->unique), server_job->function->function_name, server_job->function->function_name_size); if (gearmand_failed(ret)) { gearmand_gerror("Remove from persistent queue", ret); return ret; } } /* Job is done, remove it. */ gearman_server_job_free(server_job); break; case GEARMAN_COMMAND_SET_CLIENT_ID: gearman_server_con_set_id(server_con, (char *)(packet->arg[0]), packet->arg_size[0]); break; case GEARMAN_COMMAND_TEXT: return _server_run_text(server_con, packet); case GEARMAN_COMMAND_UNUSED: case GEARMAN_COMMAND_NOOP: case GEARMAN_COMMAND_JOB_CREATED: case GEARMAN_COMMAND_NO_JOB: case GEARMAN_COMMAND_JOB_ASSIGN: case GEARMAN_COMMAND_ECHO_RES: case GEARMAN_COMMAND_ERROR: case GEARMAN_COMMAND_STATUS_RES: case GEARMAN_COMMAND_ALL_YOURS: case GEARMAN_COMMAND_OPTION_RES: case GEARMAN_COMMAND_SUBMIT_JOB_SCHED: case GEARMAN_COMMAND_JOB_ASSIGN_UNIQ: case GEARMAN_COMMAND_JOB_ASSIGN_ALL: case GEARMAN_COMMAND_MAX: default: return _server_error_packet(server_con, "bad_command", "Command not expected"); } return GEARMAN_SUCCESS; } gearmand_error_t gearman_server_shutdown_graceful(gearman_server_st *server) { server->shutdown_graceful= true; if (server->job_count == 0) return GEARMAN_SHUTDOWN; return GEARMAN_SHUTDOWN_GRACEFUL; } gearmand_error_t gearman_server_queue_replay(gearman_server_st *server) { gearmand_error_t ret; if (server->queue._replay_fn == NULL) return GEARMAN_SUCCESS; server->state.queue_startup= true; ret= (*(server->queue._replay_fn))(server, (void *)server->queue._context, _queue_replay_add, server); server->state.queue_startup= false; return ret; } void *gearman_server_queue_context(const gearman_server_st *server) { return (void *)server->queue._context; } void gearman_server_set_queue(gearman_server_st *server, void *context, gearman_queue_add_fn *add, gearman_queue_flush_fn *flush, gearman_queue_done_fn *done, gearman_queue_replay_fn *replay) { server->queue._context= context; server->queue._add_fn= add; server->queue._flush_fn= flush; server->queue._done_fn= done; server->queue._replay_fn= replay; } /* * Private definitions */ gearmand_error_t _queue_replay_add(gearman_server_st *server, void *context __attribute__ ((unused)), const char *unique, size_t unique_size, const char *function_name, size_t function_name_size, const void *data, size_t data_size, gearmand_job_priority_t priority, int64_t when) { gearmand_error_t ret= GEARMAN_SUCCESS; (void)gearman_server_job_add(server, function_name, function_name_size, unique, unique_size, data, data_size, priority, NULL, &ret, when); if (ret != GEARMAN_SUCCESS) gearmand_gerror("gearman_server_job_add", ret); return ret; } static gearmand_error_t _server_error_packet(gearman_server_con_st *server_con, const char *error_code, const char *error_string) { return gearman_server_io_packet_add(server_con, false, GEARMAN_MAGIC_RESPONSE, GEARMAN_COMMAND_ERROR, error_code, (size_t)(strlen(error_code) + 1), error_string, (size_t)strlen(error_string), NULL); } static gearmand_error_t _server_run_text(gearman_server_con_st *server_con, gearmand_packet_st *packet) { size_t total; char *data= (char *)(char *)malloc(GEARMAN_TEXT_RESPONSE_SIZE); if (data == NULL) { gearmand_perror("malloc"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } total= GEARMAN_TEXT_RESPONSE_SIZE; data[0]= 0; if (packet->argc) { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "text command %.*s", packet->arg_size[0], packet->arg[0]); } if (packet->argc == 0) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_UNKNOWN_COMMAND, 4, "NULL"); } else if (strcasecmp("workers", (char *)(packet->arg[0])) == 0) { size_t size= 0; for (gearman_server_thread_st *thread= Server->thread_list; thread != NULL; thread= thread->next) { int error; if (! (error= pthread_mutex_lock(&thread->lock))) { for (gearman_server_con_st *con= thread->con_list; con != NULL; con= con->next) { if (con->_host == NULL) { continue; } if (size > total) { size= total; } /* Make sure we have at least GEARMAN_TEXT_RESPONSE_SIZE bytes. */ if (size + GEARMAN_TEXT_RESPONSE_SIZE > total) { char *new_data= (char *)realloc(data, total + GEARMAN_TEXT_RESPONSE_SIZE); if (new_data == NULL) { (void) pthread_mutex_unlock(&thread->lock); gearmand_perror("realloc"); gearmand_debug("free"); free(data); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } data= new_data; total+= GEARMAN_TEXT_RESPONSE_SIZE; } int sn_checked_length= snprintf(data + size, total - size, "%d %s %s :", con->con.fd, con->_host, con->id); if ((size_t)sn_checked_length > total - size || sn_checked_length < 0) { (void) pthread_mutex_unlock(&thread->lock); gearmand_debug("free"); free(data); gearmand_perror("snprintf"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } size+= (size_t)sn_checked_length; if (size > total) { continue; } for (gearman_server_worker_st *worker= con->worker_list; worker != NULL; worker= worker->con_next) { int checked_length= snprintf(data + size, total - size, " %.*s", (int)(worker->function->function_name_size), worker->function->function_name); if ((size_t)checked_length > total - size || checked_length < 0) { (void) pthread_mutex_unlock(&thread->lock); gearmand_debug("free"); free(data); gearmand_perror("snprintf"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } size+= (size_t)checked_length; if (size > total) break; } if (size > total) { continue; } int checked_length= snprintf(data + size, total - size, "\n"); if ((size_t)checked_length > total - size || checked_length < 0) { (void) pthread_mutex_unlock(&thread->lock); gearmand_debug("free"); free(data); gearmand_perror("snprintf"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } size+= (size_t)checked_length; } (void) pthread_mutex_unlock(&thread->lock); } else { errno= error; gearmand_error("pthread_mutex_lock"); assert(! "pthread_mutex_lock"); } } if (size < total) { int checked_length= snprintf(data + size, total - size, ".\n"); if ((size_t)checked_length > total - size || checked_length < 0) { gearmand_perror("snprintf"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } } } else if (strcasecmp("status", (char *)(packet->arg[0])) == 0) { size_t size= 0; gearman_server_function_st *function; for (function= Server->function_list; function != NULL; function= function->next) { if (size + GEARMAN_TEXT_RESPONSE_SIZE > total) { char *new_data= (char *)realloc(data, total + GEARMAN_TEXT_RESPONSE_SIZE); if (new_data == NULL) { gearmand_perror("realloc"); gearmand_debug("free"); free(data); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } data= new_data; total+= GEARMAN_TEXT_RESPONSE_SIZE; } int checked_length= snprintf(data + size, total - size, "%.*s\t%u\t%u\t%u\n", (int)(function->function_name_size), function->function_name, function->job_total, function->job_running, function->worker_count); if ((size_t)checked_length > total - size || checked_length < 0) { gearmand_perror("snprintf"); gearmand_debug("free"); free(data); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } size+= (size_t)checked_length; if (size > total) { size= total; } } if (size < total) { int checked_length= snprintf(data + size, total - size, ".\n"); if ((size_t)checked_length > total - size || checked_length < 0) { gearmand_perror("snprintf"); gearmand_debug("free"); free(data); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } } } else if (strcasecmp("create", (char *)(packet->arg[0])) == 0) { if (packet->argc == 3 && !strcasecmp("function", (char *)(packet->arg[1]))) { gearman_server_function_st *function; function= gearman_server_function_get(Server, (char *)(packet->arg[2]), packet->arg_size[2] -2); if (function) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_SUCCESS); } else { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_CREATE_FUNCTION, (int)packet->arg_size[2], (char *)(packet->arg[2])); } } else { // create snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_ARGS, (int)packet->arg_size[0], (char *)(packet->arg[0])); } } else if (strcasecmp("drop", (char *)(packet->arg[0])) == 0) { if (packet->argc == 3 && !strcasecmp("function", (char *)(packet->arg[1]))) { bool success= false; for (gearman_server_function_st *function= Server->function_list; function != NULL; function= function->next) { if (! strcasecmp(function->function_name, (char *)(packet->arg[2]))) { success++; if (function->worker_count == 0 && function->job_running == 0) { gearman_server_function_free(Server, function); snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_SUCCESS); } else { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, "ERR there are still connected workers or executing clients\r\n"); } break; } } if (! success) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, "ERR function not found\r\n"); gearmand_debug(data); } } else { // drop snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_ARGS, (int)packet->arg_size[0], (char *)(packet->arg[0])); } } else if (strcasecmp("maxqueue", (char *)(packet->arg[0])) == 0) { if (packet->argc == 1) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_ARGS, (int)packet->arg_size[0], (char *)(packet->arg[0])); } else { uint32_t max_queue_size[GEARMAND_JOB_PRIORITY_MAX]; for (int priority= 0; priority < GEARMAND_JOB_PRIORITY_MAX; ++priority) { const int argc= priority +2; if (packet->argc > argc) { const int parameter= atoi((char *)(packet->arg[argc])); if (parameter < 0) { max_queue_size[priority]= 0; } else { max_queue_size[priority]= (uint32_t)parameter; } } else { max_queue_size[priority]= GEARMAN_DEFAULT_MAX_QUEUE_SIZE; } } /* To preserve the existing behavior of maxqueue, ensure that the one-parameter invocation is applied to all priorities. */ if (packet->argc <= 3) { for (int priority= 1; priority < GEARMAND_JOB_PRIORITY_MAX; ++priority) { max_queue_size[priority]= max_queue_size[0]; } } for (gearman_server_function_st *function= Server->function_list; function != NULL; function= function->next) { if (strlen((char *)(packet->arg[1])) == function->function_name_size && (memcmp(packet->arg[1], function->function_name, function->function_name_size) == 0)) { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Applying queue limits to %s", function->function_name); memcpy(function->max_queue_size, max_queue_size, sizeof(uint32_t) * GEARMAND_JOB_PRIORITY_MAX); } } snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_SUCCESS); } } else if (strcasecmp("getpid", (char *)(packet->arg[0])) == 0) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, "OK %d\n", (int)getpid()); } else if (strcasecmp("shutdown", (char *)(packet->arg[0])) == 0) { if (packet->argc == 1) { Server->shutdown= true; snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_SUCCESS); } else if (packet->argc == 2 && strcasecmp("graceful", (char *)(packet->arg[1])) == 0) { Server->shutdown_graceful= true; snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_SUCCESS); } else { // shutdown snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_ARGS, (int)packet->arg_size[0], (char *)(packet->arg[0])); } } else if (strcasecmp("verbose", (char *)(packet->arg[0])) == 0) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, "OK %s\n", gearmand_verbose_name(Gearmand()->verbose)); } else if (strcasecmp("version", (char *)(packet->arg[0])) == 0) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, "OK %s\n", PACKAGE_VERSION); } else { gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Failed to find command %.*s(%llu)", packet->arg_size[0], packet->arg[0], (unsigned long long)packet->arg_size[0]); snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_UNKNOWN_COMMAND, (int)packet->arg_size[0], (char *)(packet->arg[0])); } gearman_server_packet_st *server_packet= gearman_server_packet_create(server_con->thread, false); if (server_packet == NULL) { gearmand_debug("free"); free(data); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } gearmand_packet_init(&(server_packet->packet), GEARMAN_MAGIC_TEXT, GEARMAN_COMMAND_TEXT); server_packet->packet.magic= GEARMAN_MAGIC_TEXT; server_packet->packet.command= GEARMAN_COMMAND_TEXT; server_packet->packet.options.complete= true; server_packet->packet.options.free_data= true; if (data[0] == 0) { snprintf(data, GEARMAN_TEXT_RESPONSE_SIZE, TEXT_ERROR_INTERNAL_ERROR); } server_packet->packet.data= data; server_packet->packet.data_size= strlen(data); int error; if (! (error= pthread_mutex_lock(&server_con->thread->lock))) { GEARMAN_FIFO_ADD(server_con->io_packet, server_packet,); (void) pthread_mutex_unlock(&server_con->thread->lock); } else { errno= error; gearmand_perror("pthread_mutex_lock"); assert(!"pthread_mutex_lock"); } gearman_server_con_io_add(server_con); return GEARMAN_SUCCESS; } static gearmand_error_t _server_queue_work_data(gearman_server_job_st *server_job, gearmand_packet_st *packet, gearman_command_t command) { gearman_server_client_st *server_client; uint8_t *data; gearmand_error_t ret; for (server_client= server_job->client_list; server_client; server_client= server_client->job_next) { if (command == GEARMAN_COMMAND_WORK_EXCEPTION && !(server_client->con->is_exceptions)) { gearmand_debug("Dropping GEARMAN_COMMAND_WORK_EXCEPTION"); continue; } else if (command == GEARMAN_COMMAND_WORK_EXCEPTION) { gearmand_debug("Sending GEARMAN_COMMAND_WORK_EXCEPTION"); } if (packet->data_size > 0) { if (packet->options.free_data && server_client->job_next == NULL) { data= (uint8_t *)(packet->data); packet->options.free_data= false; } else { data= (uint8_t *)malloc(packet->data_size); if (data == NULL) { gearmand_perror("malloc"); return GEARMAN_MEMORY_ALLOCATION_FAILURE; } memcpy(data, packet->data, packet->data_size); } } else { data= NULL; } ret= gearman_server_io_packet_add(server_client->con, true, GEARMAN_MAGIC_RESPONSE, command, packet->arg[0], packet->arg_size[0], data, packet->data_size, NULL); if (ret != GEARMAN_SUCCESS) { gearmand_gerror("gearman_server_io_packet_add", ret); return ret; } } return GEARMAN_SUCCESS; }