123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493 |
- /* 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 Gearman State Definitions
- */
- #include <libgearman/common.h>
- #include <libgearman/connection.h>
- #include <libgearman/packet.hpp>
- #include <libgearman/universal.hpp>
- #include <cassert>
- #include <cerrno>
- #include <cstdarg>
- #include <cstdio>
- #include <cstdlib>
- #include <cstring>
- void gearman_universal_initialize(gearman_universal_st &self, const gearman_universal_options_t *options)
- {
- { // Set defaults on all options.
- self.options.dont_track_packets= false;
- self.options.non_blocking= false;
- self.options.stored_non_blocking= false;
- }
- if (options)
- {
- while (*options != GEARMAN_MAX)
- {
- /**
- @note Check for bad value, refactor gearman_add_options().
- */
- gearman_universal_add_options(self, *options);
- options++;
- }
- }
- self.verbose= GEARMAN_VERBOSE_NEVER;
- self.con_count= 0;
- self.packet_count= 0;
- self.pfds_size= 0;
- self.sending= 0;
- self.timeout= -1;
- self.con_list= NULL;
- self.packet_list= NULL;
- self.pfds= NULL;
- self.log_fn= NULL;
- self.log_context= NULL;
- self.workload_malloc_fn= NULL;
- self.workload_malloc_context= NULL;
- self.workload_free_fn= NULL;
- self.workload_free_context= NULL;
- self.error.rc= GEARMAN_SUCCESS;
- self.error.last_errno= 0;
- self.error.last_error[0]= 0;
- }
- void gearman_universal_clone(gearman_universal_st &destination, const gearman_universal_st &source)
- {
- gearman_universal_initialize(destination);
- (void)gearman_universal_set_option(destination, GEARMAN_NON_BLOCKING, source.options.non_blocking);
- (void)gearman_universal_set_option(destination, GEARMAN_DONT_TRACK_PACKETS, source.options.dont_track_packets);
- destination.timeout= source.timeout;
- for (gearman_connection_st *con= source.con_list; con; con= con->next)
- {
- if (gearman_connection_clone(&destination, NULL, con) == NULL)
- {
- gearman_universal_free(destination);
- return;
- }
- }
- }
- void gearman_universal_free(gearman_universal_st &universal)
- {
- gearman_free_all_cons(&universal);
- gearman_free_all_packets(universal);
- if (universal.pfds)
- {
- // created realloc()
- free(universal.pfds);
- }
- }
- gearman_return_t gearman_universal_set_option(gearman_universal_st &self, gearman_universal_options_t option, bool value)
- {
- switch (option)
- {
- case GEARMAN_NON_BLOCKING:
- self.options.non_blocking= value;
- break;
- case GEARMAN_DONT_TRACK_PACKETS:
- self.options.dont_track_packets= value;
- break;
- case GEARMAN_MAX:
- default:
- return GEARMAN_INVALID_COMMAND;
- }
- return GEARMAN_SUCCESS;
- }
- int gearman_universal_timeout(gearman_universal_st &self)
- {
- return self.timeout;
- }
- void gearman_universal_set_timeout(gearman_universal_st &self, int timeout)
- {
- self.timeout= timeout;
- }
- void gearman_set_log_fn(gearman_universal_st &self, gearman_log_fn *function,
- void *context, gearman_verbose_t verbose)
- {
- self.log_fn= function;
- self.log_context= context;
- self.verbose= verbose;
- }
- void gearman_set_workload_malloc_fn(gearman_universal_st *universal,
- gearman_malloc_fn *function,
- void *context)
- {
- universal->workload_malloc_fn= function;
- universal->workload_malloc_context= context;
- }
- void gearman_set_workload_free_fn(gearman_universal_st *universal,
- gearman_free_fn *function,
- void *context)
- {
- universal->workload_free_fn= function;
- universal->workload_free_context= context;
- }
- void gearman_free_all_cons(gearman_universal_st *universal)
- {
- while (universal->con_list != NULL)
- gearman_connection_free(universal->con_list);
- }
- gearman_return_t gearman_flush_all(gearman_universal_st *universal)
- {
- for (gearman_connection_st *con= universal->con_list; con != NULL; con= con->next)
- {
- if (con->events & POLLOUT)
- continue;
- gearman_return_t ret= gearman_connection_flush(con);
- if (gearman_failed(ret) and ret != GEARMAN_IO_WAIT)
- return ret;
- }
- return GEARMAN_SUCCESS;
- }
- gearman_return_t gearman_wait(gearman_universal_st *universal)
- {
- struct pollfd *pfds;
- nfds_t x;
- int ret;
- gearman_return_t gret;
- if (universal->pfds_size < universal->con_count)
- {
- pfds= static_cast<pollfd*>(realloc(universal->pfds, universal->con_count * sizeof(struct pollfd)));
- if (pfds == NULL)
- {
- gearman_perror(*universal, "pollfd realloc");
- return GEARMAN_MEMORY_ALLOCATION_FAILURE;
- }
- universal->pfds= pfds;
- universal->pfds_size= universal->con_count;
- }
- else
- {
- pfds= universal->pfds;
- }
- x= 0;
- for (gearman_connection_st *con= universal->con_list; con != NULL; con= con->next)
- {
- if (con->events == 0)
- continue;
- pfds[x].fd= con->fd;
- pfds[x].events= con->events;
- pfds[x].revents= 0;
- x++;
- }
- if (x == 0)
- {
- gearman_error(universal, GEARMAN_NO_ACTIVE_FDS, "no active file descriptors");
- return GEARMAN_NO_ACTIVE_FDS;
- }
- while (1)
- {
- ret= poll(pfds, x, universal->timeout);
- if (ret == -1)
- {
- if (errno == EINTR)
- continue;
- gearman_perror(*universal, "poll");
- return GEARMAN_ERRNO;
- }
- break;
- }
- if (ret == 0)
- {
- gearman_error(universal, GEARMAN_TIMEOUT, "timeout reached");
- return GEARMAN_TIMEOUT;
- }
- x= 0;
- for (gearman_connection_st *con= universal->con_list; con != NULL; con= con->next)
- {
- if (con->events == 0)
- continue;
- gret= gearman_connection_set_revents(con, pfds[x].revents);
- if (gret != GEARMAN_SUCCESS)
- return gret;
- x++;
- }
- return GEARMAN_SUCCESS;
- }
- gearman_connection_st *gearman_ready(gearman_universal_st *universal)
- {
- gearman_connection_st *con;
- /* We can't keep universal between calls since connections may be removed during
- processing. If this list ever gets big, we may want something faster. */
- for (con= universal->con_list; con != NULL; con= con->next)
- {
- if (con->options.ready)
- {
- con->options.ready= false;
- return con;
- }
- }
- return NULL;
- }
- /**
- @note _push_blocking is only used for echo (and should be fixed
- when tricky flip/flop in IO is fixed).
- */
- static inline void _push_blocking(gearman_universal_st *universal, bool *orig_block_universal)
- {
- *orig_block_universal= universal->options.non_blocking;
- universal->options.non_blocking= false;
- }
- static inline void _pop_non_blocking(gearman_universal_st *universal, bool orig_block_universal)
- {
- universal->options.non_blocking= orig_block_universal;
- }
- gearman_return_t gearman_echo(gearman_universal_st *universal,
- const void *workload,
- size_t workload_size)
- {
- gearman_connection_st *con;
- gearman_packet_st packet;
- gearman_return_t ret;
- bool orig_block_universal;
- ret= gearman_packet_create_args(*universal, &packet, GEARMAN_MAGIC_REQUEST,
- GEARMAN_COMMAND_ECHO_REQ,
- &workload, &workload_size, 1);
- if (gearman_failed(ret))
- {
- return ret;
- }
- _push_blocking(universal, &orig_block_universal);
- for (con= universal->con_list; con != NULL; con= con->next)
- {
- gearman_packet_st *packet_ptr;
- ret= gearman_connection_send(con, &packet, true);
- if (gearman_failed(ret))
- {
- goto exit;
- }
- packet_ptr= gearman_connection_recv(con, &(con->packet), &ret, true);
- if (gearman_failed(ret))
- {
- goto exit;
- }
- assert(packet_ptr);
- if (con->packet.data_size != workload_size ||
- memcmp(workload, con->packet.data, workload_size))
- {
- gearman_packet_free(&(con->packet));
- gearman_error(universal, GEARMAN_ECHO_DATA_CORRUPTION, "corruption during echo");
- ret= GEARMAN_ECHO_DATA_CORRUPTION;
- goto exit;
- }
- gearman_packet_free(&(con->packet));
- }
- ret= GEARMAN_SUCCESS;
- exit:
- gearman_packet_free(&packet);
- _pop_non_blocking(universal, orig_block_universal);
- return ret;
- }
- bool gearman_request_option(gearman_universal_st &universal,
- gearman_string_t &option)
- {
- gearman_connection_st *con;
- gearman_packet_st packet;
- gearman_return_t ret;
- bool orig_block_universal;
- const void *args[]= { gearman_c_str(option) };
- size_t args_size[]= { gearman_size(option) };
- ret= gearman_packet_create_args(universal, &packet, GEARMAN_MAGIC_REQUEST,
- GEARMAN_COMMAND_OPTION_REQ,
- args, args_size, 1);
- if (gearman_failed(ret))
- {
- gearman_error(&universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "gearman_packet_create_args()");
- return ret;
- }
- _push_blocking(&universal, &orig_block_universal);
- for (con= universal.con_list; con != NULL; con= con->next)
- {
- gearman_packet_st *packet_ptr;
- ret= gearman_connection_send(con, &packet, true);
- if (gearman_failed(ret))
- {
- gearman_packet_free(&(con->packet));
- goto exit;
- }
- packet_ptr= gearman_connection_recv(con, &(con->packet), &ret, true);
- if (gearman_failed(ret))
- {
- gearman_packet_free(&(con->packet));
- goto exit;
- }
- assert(packet_ptr);
- if (packet_ptr->command == GEARMAN_COMMAND_ERROR)
- {
- gearman_packet_free(&(con->packet));
- gearman_error(&universal, GEARMAN_INVALID_ARGUMENT, "invalid server option");
- ret= GEARMAN_INVALID_ARGUMENT;;
- goto exit;
- }
- gearman_packet_free(&(con->packet));
- }
- ret= GEARMAN_SUCCESS;
- exit:
- gearman_packet_free(&packet);
- _pop_non_blocking(&universal, orig_block_universal);
- return gearman_success(ret);
- }
- void gearman_free_all_packets(gearman_universal_st &universal)
- {
- while (universal.packet_list)
- gearman_packet_free(universal.packet_list);
- }
- /*
- * Local Definitions
- */
- void gearman_universal_set_error(gearman_universal_st *universal,
- gearman_return_t rc,
- const char *function,
- const char *format, ...)
- {
- size_t size;
- char log_buffer[GEARMAN_MAX_ERROR_SIZE];
- char *ptr= log_buffer;
- va_list args;
- universal->error.rc= rc;
- if (rc != GEARMAN_ERRNO)
- {
- universal->error.last_errno= 0;
- }
- size= strlen(gearman_strerror(rc));
- ptr= static_cast<char *>(memcpy(ptr, gearman_strerror(rc), size));
- ptr+= size;
- ptr[0]= '-';
- size++;
- ptr++;
- ptr[0]= '>';
- size++;
- ptr++;
- size= strlen(function);
- ptr= static_cast<char *>(memcpy(ptr, static_cast<void*>(const_cast<char *>(function)), size));
- ptr+= size;
- ptr[0]= ':';
- size++;
- ptr++;
- ptr[0]= ' ';
- size++;
- ptr++;
- va_start(args, format);
- size+= size_t(vsnprintf(ptr, GEARMAN_MAX_ERROR_SIZE - size, format, args));
- va_end(args);
- if (universal->log_fn == NULL)
- {
- if (size >= GEARMAN_MAX_ERROR_SIZE)
- size= GEARMAN_MAX_ERROR_SIZE - 1;
- memcpy(universal->error.last_error, log_buffer, size + 1);
- }
- else
- {
- universal->log_fn(log_buffer, GEARMAN_VERBOSE_FATAL,
- static_cast<void *>(universal->log_context));
- }
- }
- void gearman_universal_set_perror(const char *position, gearman_universal_st &self, const char *message)
- {
- self.error.rc= GEARMAN_ERRNO;
- self.error.last_errno= errno;
- const char *errmsg_ptr;
- char errmsg[GEARMAN_MAX_ERROR_SIZE];
- errmsg[0]= 0;
- #ifdef STRERROR_R_CHAR_P
- errmsg_ptr= strerror_r(self.error.last_errno, errmsg, sizeof(errmsg));
- #else
- strerror_r(self.error.last_errno, errmsg, sizeof(errmsg));
- errmsg_ptr= errmsg;
- #endif
- char final[GEARMAN_MAX_ERROR_SIZE];
- snprintf(final, sizeof(final), "%s(%s)", message, errmsg_ptr);
- gearman_universal_set_error(&self, GEARMAN_ERRNO, position, final);
- }
|