<commit_before>/* Gearman server and library
 * Copyright (C) 2008 <NAME>, <NAME>
 * All rights reserved.
 *
 * Use and distribution licensed under the BSD license.  See
 * the COPYING file in the parent directory for full text.
 */

/**
 * @file
 * @brief Gearmand Thread Definitions
 */

#include "common.h"

/*
 * Private declarations
 */

/**
 * @addtogroup gearmand_thread private Private Gearmand Thread Functions
 * @ingroup gearmand_thread
 * @{
 */

static void *_thread(void *data);
static void _log(gearman_server_thread_st *thread __attribute__ ((unused)),
                 uint8_t verbose, const char *line, void *arg);
static void _run(gearman_server_thread_st *thread __attribute__ ((unused)),
                 void *fn_arg);

static gearman_return_t _wakeup_init(gearmand_thread_st *thread);
static void _wakeup_close(gearmand_thread_st *thread);
static void _wakeup_clear(gearmand_thread_st *thread);
static void _wakeup_event(int fd, short events __attribute__ ((unused)),
                          void *arg);
static void _clear_events(gearmand_thread_st *thread);

/** @} */

/*
 * Public definitions
 */

gearman_return_t gearmand_thread_create(gearmand_st *gearmand)
{
  gearmand_thread_st *thread;
  gearman_return_t ret;
  int pthread_ret;

  thread= malloc(sizeof(gearmand_thread_st));
  if (thread == NULL)
  {
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create", "malloc")
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
  }

  memset(thread, 0, sizeof(gearmand_thread_st));

  thread->gearmand= gearmand;

  if (gearman_server_thread_create(&(gearmand->server),
                                   &(thread->server_thread)) == NULL)
  {
    free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                       "gearman_server_thread_create:NULL")
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
  }

  gearman_server_thread_set_log(&(thread->server_thread), _log, thread);
  gearman_server_thread_set_event_watch(&(thread->server_thread),
                                        gearmand_con_watch, NULL);

  GEARMAN_LIST_ADD(gearmand->thread, thread,)

  /* If we have no threads, we still create a fake thread that uses the main
     libevent instance. Otherwise create a libevent instance for each thread. */
  if (gearmand->threads == 0)
    thread->base= gearmand->base;
  else
  {
    GEARMAND_LOG(gearmand, 1, "Initializing libevent for IO thread")

    thread->base= event_base_new();
    if (thread->base == NULL)
    {
      gearmand_thread_free(thread);
      GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                         "event_base_new:NULL")
      return GEARMAN_EVENT;
    }
  }

  ret= _wakeup_init(thread);
  if (ret != GEARMAN_SUCCESS)
  {
    gearmand_thread_free(thread);
    return ret;
  }

  /* If we are not running multi-threaded, just return the thread context. */
  if (gearmand->threads == 0)
    return GEARMAN_SUCCESS;

  thread->count= gearmand->thread_count;

  pthread_ret= pthread_mutex_init(&(thread->lock), NULL);
  if (pthread_ret != 0)
  {
    thread->count= 0;
    gearmand_thread_free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                       "pthread_mutex_init:%d", pthread_ret)
    return GEARMAN_PTHREAD;
  }

  thread->options|= GEARMAND_THREAD_LOCK;

  gearman_server_thread_set_run(&(thread->server_thread), _run, thread);

  pthread_ret= pthread_create(&(thread->id), NULL, _thread, thread);
  if (pthread_ret != 0)
  {
    thread->count= 0;
    gearmand_thread_free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create", "pthread_create:%d",
                       pthread_ret)
    return GEARMAN_PTHREAD;
  }

  GEARMAND_LOG(gearmand, 1, "Thread %u created", thread->count)

  return GEARMAN_SUCCESS;
}

void gearmand_thread_free(gearmand_thread_st *thread)
{
  gearmand_con_st *dcon;

  if (thread->gearmand->threads && thread->count > 0)
  {
    GEARMAND_LOG(thread->gearmand, 1, "Shutting down thread %u", thread->count)

    gearmand_thread_wakeup(thread, GEARMAND_WAKEUP_SHUTDOWN);
    (void) pthread_join(thread->id, NULL);
  }

  if (thread->options & GEARMAND_THREAD_LOCK)
    (void) pthread_mutex_destroy(&(thread->lock));

  _wakeup_close(thread);

  assert(thread->dcon_list == NULL);

  while (thread->dcon_add_list != NULL)
  {
    dcon= thread->dcon_add_list;
    thread->dcon_add_list= dcon->next;
    close(dcon->fd);
    free(dcon);
  }

  while (thread->free_dcon_list != NULL)
  {
    dcon= thread->free_dcon_list;
    thread->free_dcon_list= dcon->next;
    free(dcon);
  }

  gearman_server_thread_free(&(thread->server_thread));

  GEARMAN_LIST_DEL(thread->gearmand->thread, thread,)

  if (thread->gearmand->threads > 0)
  {
    if (thread->base != NULL)<commit_msg>More cleanup.<commit_after>/* Gearman server and library
 * Copyright (C) 2008 <NAME>, <NAME>
 * All rights reserved.
 *
 * Use and distribution licensed under the BSD license.  See
 * the COPYING file in the parent directory for full text.
 */

/**
 * @file
 * @brief Gearmand Thread Definitions
 */

#include "common.h"

/*
 * Private declarations
 */

/**
 * @addtogroup gearmand_thread private Private Gearmand Thread Functions
 * @ingroup gearmand_thread
 * @{
 */

static void *_thread(void *data);
static void _log(gearman_server_thread_st *thread __attribute__ ((unused)),
                 uint8_t verbose, const char *line, void *arg);
static void _run(gearman_server_thread_st *thread __attribute__ ((unused)),
                 void *fn_arg);

static gearman_return_t _wakeup_init(gearmand_thread_st *thread);
static void _wakeup_close(gearmand_thread_st *thread);
static void _wakeup_clear(gearmand_thread_st *thread);
static void _wakeup_event(int fd, short events __attribute__ ((unused)),
                          void *arg);
static void _clear_events(gearmand_thread_st *thread);

/** @} */

/*
 * Public definitions
 */

gearman_return_t gearmand_thread_create(gearmand_st *gearmand)
{
  gearmand_thread_st *thread;
  gearman_return_t ret;
  int pthread_ret;

  thread= malloc(sizeof(gearmand_thread_st));
  if (thread == NULL)
  {
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create", "malloc")
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
  }

  memset(thread, 0, sizeof(gearmand_thread_st));

  thread->gearmand= gearmand;

  if (gearman_server_thread_create(&(gearmand->server),
                                   &(thread->server_thread)) == NULL)
  {
    free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                       "gearman_server_thread_create:NULL")
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
  }

  gearman_server_thread_set_log(&(thread->server_thread), _log, thread);
  gearman_server_thread_set_event_watch(&(thread->server_thread),
                                        gearmand_con_watch, NULL);

  GEARMAN_LIST_ADD(gearmand->thread, thread,)

  /* If we have no threads, we still create a fake thread that uses the main
     libevent instance. Otherwise create a libevent instance for each thread. */
  if (gearmand->threads == 0)
    thread->base= gearmand->base;
  else
  {
    GEARMAND_LOG(gearmand, 1, "Initializing libevent for IO thread")

    thread->base= event_base_new();
    if (thread->base == NULL)
    {
      gearmand_thread_free(thread);
      GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                         "event_base_new:NULL")
      return GEARMAN_EVENT;
    }
  }

  ret= _wakeup_init(thread);
  if (ret != GEARMAN_SUCCESS)
  {
    gearmand_thread_free(thread);
    return ret;
  }

  /* If we are not running multi-threaded, just return the thread context. */
  if (gearmand->threads == 0)
    return GEARMAN_SUCCESS;

  thread->count= gearmand->thread_count;

  pthread_ret= pthread_mutex_init(&(thread->lock), NULL);
  if (pthread_ret != 0)
  {
    thread->count= 0;
    gearmand_thread_free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create",
                       "pthread_mutex_init:%d", pthread_ret)
    return GEARMAN_PTHREAD;
  }

  thread->options|= GEARMAND_THREAD_LOCK;

  gearman_server_thread_set_run(&(thread->server_thread), _run, thread);

  pthread_ret= pthread_create(&(thread->id), NULL, _thread, thread);
  if (pthread_ret != 0)
  {
    thread->count= 0;
    gearmand_thread_free(thread);
    GEARMAND_ERROR_SET(gearmand, "gearmand_thread_create", "pthread_create:%d",
                       pthread_ret)
    return GEARMAN_PTHREAD;
  }

  GEARMAND_LOG(gearmand, 1, "Thread %u created", thread->count)

  return GEARMAN_SUCCESS;
}

void gearmand_thread_free(gearmand_thread_st *thread)
{
  gearmand_con_st *dcon;

  if (thread->gearmand->threads && thread->count > 0)
  {
    GEARMAND_LOG(thread->gearmand, 1, "Shutting down thread %u", thread->count)

    gearmand_thread_wakeup(thread, GEARMAND_WAKEUP_SHUTDOWN);
    (void) pthread_join(thread->id, NULL);
  }

  if (thread->options & GEARMAND_THREAD_LOCK)
    (void) pthread_mutex_destroy(&(thread->lock));

  _wakeup_close(thread);

  while (thread->dcon_list != NULL)
    gearmand_con_free(thread->dcon_list);

  while (thread->dcon_add_list != NULL)
  {
    dcon= thread->dcon_add_list;
    thread->dcon_add_list= dcon->next;
    close(dcon->fd);
    free(dcon);
  }

  while (thread->free_dcon_list != NULL)
  {
    dcon= thread->free_dcon_list;
    thread->free_dcon_list= dcon->next;
    free(dcon);
  }

  gearman_server_thread_free(&(thread->server_thread));

  GEARMAN_LIST_DEL(thread->gearmand->thread, thread,)

  if (thread->gearmand->threads > 0)
  {
    if (thread->base != NULL)
{'fraction_non_alphanumeric': 0.1151794120723925, 'fraction_numerical': 0.0035568574118631656, 'mean_word_length': 3.7000983284169124, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 102, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7433265', 'n_tokens_mistral': 3374, 'n_tokens_neox': 3259, 'n_words': 642}