1

#include "ctx.h"

     2

#include "error.h"

     3

#include "shared/log.h"

     4

     5

#include <stdlib.h>

     6

#include <signal.h>

     7

#include <assert.h>

     8

#include <stdio.h> // for perror

     9

    10

/**

    11

 * Wrapper around pthread_mutex_unlock for use with pthread_cleanup_push

    12

 */

    13

static void pt_mutex_unlock (void *arg)

    14

{

    15

    pthread_mutex_t *mutex = arg;

    16

    17

    log_debug("cleanup");

    18

    19

    assert(!pthread_mutex_unlock(mutex));

    20

}

    21

    22

/**

    23

 * Enqueue the given piece of work

    24

 *

    25

 * This function always succeeds.

    26

 */

    27

static void pt_work_enqueue (struct pt_ctx *ctx, struct pt_work *work)

    28

{

    29

    // acquire

    30

    assert(!pthread_mutex_lock(&ctx->work_mutex));

    31

    32

    // enqueue work

    33

    TAILQ_INSERT_TAIL(&ctx->work, work, ctx_work);

    34

    35

    // if there's a thread waiting, wake one up. Otherwise, some thread will find it once it finishes its current work

    36

    assert(!pthread_cond_signal(&ctx->work_cond));

    37

    38

    // release 

    39

    assert(!pthread_mutex_unlock(&ctx->work_mutex));

    40

}

    41

    42

/**

    43

 * Dequeue a piece of work, returning true while there was still work to do

    44

 */

    45

static bool pt_work_dequeue (struct pt_ctx *ctx, struct pt_work **work_ptr)

    46

{

    47

    bool ret;

    48

    49

    // acquire, cancel-safe

    50

    pthread_cleanup_push(pt_mutex_unlock, &ctx->work_mutex);

    51

    assert(!pthread_mutex_lock(&ctx->work_mutex));

    52

    53

    // wait for work

    54

    while (ctx->running && TAILQ_EMPTY(&ctx->work))

    55

        // we can expect to get pthread_cancel'd here

    56

        assert(!pthread_cond_wait(&ctx->work_cond, &ctx->work_mutex));

    57

    58

    // still got work?

    59

    if (!TAILQ_EMPTY(&ctx->work)) {

    60

        // pop work

    61

        *work_ptr = TAILQ_FIRST(&ctx->work);

    62

        TAILQ_REMOVE(&ctx->work, *work_ptr, ctx_work);

    63

    64

        log_debug("got work");

    65

    66

        ret = true;

    67

    68

    } else {

    69

        assert(!ctx->running);

    70

    71

        // work empty, idle

    72

        assert(!pthread_cond_signal(&ctx->idle_cond));

    73

    74

        log_debug("idle/dead");

    75

    76

        // no more work

    77

        ret = false;

    78

    }

    79

    80

    // release

    81

    pthread_cleanup_pop(true);

    82

    83

    return ret;

    84

}

    85

    86

/**

    87

 * Wait for work queue to empty and workers to finish

    88

 */

    89

static void pt_work_shutdown (struct pt_ctx *ctx)

    90

{

    91

    // acquire

    92

    assert(!pthread_mutex_lock(&ctx->work_mutex));

    93

    94

    // indicate to terminate

    95

    ctx->running = false;

    96

    97

    // wake up any idle workers

    98

    assert(!pthread_cond_broadcast(&ctx->work_cond));

    99

   100

    // wait for it to drain...

   101

    while (!TAILQ_EMPTY(&ctx->work))

   102

        assert(!pthread_cond_wait(&ctx->idle_cond, &ctx->work_mutex));

   103

   104

    // release

   105

    assert(!pthread_mutex_unlock(&ctx->work_mutex));

   106

}

   107

   108

/**

   109

 * Execute a piece of work

   110

 */

   111

static void pt_work_execute (struct pt_work *work)

   112

{

   113

    work->func(work->arg);

   114

}

   115

   116

/**

   117

 * Release work state once done

   118

 */

   119

static void pt_work_release (struct pt_work *work)

   120

{

   121

    free(work);

   122

}

   123

   124

/**

   125

 * Worker thread entry point

   126

 */

   127

static void* pt_thread_main (void *arg)

   128

{

   129

    struct pt_thread *thread = arg;

   130

    struct pt_work *work;

   131

   132

    log_debug("start");

   133

   134

    // if only life were so simple...

   135

    while (pt_work_dequeue(thread->ctx, &work)) {

   136

        log_debug("work_execute");

   137

        pt_work_execute(work);

   138

        pt_work_release(work);

   139

        log_debug("work_done");

   140

    }

   141

   142

    log_debug("exit");

   143

   144

    return NULL;

   145

}

   146

   147

/**

   148

 * Shut down a pt_thread, waiting for it to finish.

   149

 *

   150

 * Does not remove the thread from the pool or release the pt_thread.

   151

 */

   152

static void pt_thread_shutdown (struct pt_thread *thread)

   153

{

   154

    // wait for it to finish

   155

    if (pthread_join(thread->tid, NULL))

   156

        perror("pthread_join");

   157

   158

    // mark

   159

    thread->tid = 0;

   160

}

   161

   162

/**

   163

 * Release pt_thread state, aborting thread if running.

   164

 *

   165

 * This is guaranteed to remove the thread from the ctx_threads if it was added.

   166

 */

   167

static void pt_thread_destroy (struct pt_thread *thread)

   168

{

   169

    if (thread->tid) {

   170

        // detach

   171

        if (pthread_detach(thread->tid))

   172

            perror("pthread_detach");

   173

   174

        // abort thread

   175

        if (pthread_kill(thread->tid, SIGTERM))

   176

            perror("pthread_detach");

   177

   178

    }

   179

   180

    // remove from pool

   181

    TAILQ_REMOVE(&thread->ctx->threads, thread, ctx_threads);

   182

   183

    free(thread);

   184

}

   185

   186

   187

/**

   188

 * Start a new thread and add it to the thread pool

   189

 */

   190

static int pt_ctx_add_thread (struct pt_ctx *ctx)

   191

{

   192

    struct pt_thread *thread;

   193

    int err;

   194

   195

    // alloc

   196

    if ((thread = calloc(1, sizeof(*thread))) == NULL)

   197

        return -PT_ERR_MEM;

   198

   199

    // init

   200

    thread->ctx = ctx;

   201

   202

    // start thread, default attrs

   203

    if (pthread_create(&thread->tid, NULL, pt_thread_main, thread))

   204

        JUMP_SET_ERROR(err, PT_ERR_PTHREAD_CREATE);

   205

   206

    // add to pool

   207

    TAILQ_INSERT_TAIL(&ctx->threads, thread, ctx_threads);

   208

   209

    // ok

   210

    return 0;

   211

   212

error:

   213

    // drop, don't try and remove from tailq...

   214

    free(thread);

   215

   216

    return err;

   217

}

   218

   219

int pt_ctx_new (struct pt_ctx **ctx_ptr, int threads)

   220

{

   221

    struct pt_ctx *ctx;

   222

    int err;

   223

   224

    // alloc

   225

    if ((ctx = calloc(1, sizeof(*ctx))) == NULL)

   226

        return -PT_ERR_MEM;

   227

   228

    // init

   229

    TAILQ_INIT(&ctx->threads);

   230

    TAILQ_INIT(&ctx->work);

   231

    pthread_mutex_init(&ctx->work_mutex, NULL);

   232

    pthread_cond_init(&ctx->work_cond, NULL);

   233

    pthread_cond_init(&ctx->idle_cond, NULL);

   234

    ctx->running = true;

   235

   236

    // start threadpool

   237

    for (int i = 0; i < threads; i++) {

   238

        if ((err = pt_ctx_add_thread(ctx)))

   239

            JUMP_ERROR(err);

   240

    }

   241

   242

    // ok

   243

    *ctx_ptr = ctx;

   244

   245

    return 0;

   246

   247

error:

   248

    // cleanup

   249

    pt_ctx_destroy(ctx);

   250

   251

    return err;

   252

}

   253

   254

int pt_ctx_work (struct pt_ctx *ctx, pt_work_func func, void *arg)

   255

{

   256

    struct pt_work *work;

   257

   258

    // check state

   259

    // XXX: this is kind of retarded, because pt_ctx_shutdown/work should only be called from the same thread...

   260

    if (!ctx->running)

   261

        RETURN_ERROR(PT_ERR_CTX_SHUTDOWN);

   262

   263

    // alloc

   264

    if ((work = calloc(1, sizeof(*work))) == NULL)

   265

        RETURN_ERROR(PT_ERR_MEM);

   266

   267

    // init

   268

    work->func = func;

   269

    work->arg = arg;

   270

   271

    // enqueue

   272

    pt_work_enqueue(ctx, work);

   273

   274

    // ok

   275

    return 0;

   276

}

   277

   278

int pt_ctx_shutdown (struct pt_ctx *ctx)

   279

{

   280

    struct pt_thread *thread;

   281

   282

    // finish work

   283

    pt_work_shutdown(ctx);

   284

   285

    // shut down each thread

   286

    TAILQ_FOREACH(thread, &ctx->threads, ctx_threads)

   287

        // XXX: handle errors of some kind?

   288

        pt_thread_shutdown(thread);

   289

   290

    // then drop

   291

    pt_ctx_destroy(ctx);

   292

   293

    return 0;

   294

}

   295

   296

void pt_ctx_destroy (struct pt_ctx *ctx)

   297

{

   298

    struct pt_thread *thread;

   299

   300

    // kill threads

   301

    while ((thread = TAILQ_FIRST(&ctx->threads)))

   302

        pt_thread_destroy(thread);

   303

   304

    // destroy mutex/conds

   305

    if (pthread_cond_destroy(&ctx->idle_cond))

   306

        perror("pthread_cond_destroy(idle_cond)");

   307

   308

    if (pthread_cond_destroy(&ctx->work_cond))

   309

        perror("pthread_cond_destroy(work_cond)");

   310

   311

    if (pthread_mutex_destroy(&ctx->work_mutex))

   312

        perror("pthread_mutex_destroy(work_mutex)");

   313

   314

   315

    free(ctx);

   316

}

   317