1

#include "ctx.h"

     2

#include "error.h"

     3

     4

#include <stdlib.h>

     5

#include <signal.h>

     6

#include <assert.h>

     7

#include <stdio.h> // for perror

     8

    12

static void pt_mutex_unlock (void *arg)

    13

{

    14

    pthread_mutex_t *mutex = arg;

    15

    16

    assert(!pthread_mutex_unlock(mutex));

    17

}

    18

    19

/**

    20

 * Enqueue the given piece of work

    21

 *

    22

 * This function always succeeds.

    23

 */

    24

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

    25

{

    26

    // acquire

    27

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

    28

    29

    // enqueue work

    30

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

    31

    32

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

    33

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

    34

    35

    // release 

    36

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

    37

}

    38

    39

/**

    40

 * Dequeue a piece of work

    41

 */

    42

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

    43

{

    44

    // acquire, cancel-safe

    45

    pthread_cleanup_push(pt_mutex_unlock, &ctx->work_mutex);

    46

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

    47

    48

    // idle?

    49

    if (TAILQ_EMPTY(&ctx->work))

    50

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

    51

    52

    // wait for work

    53

    while (TAILQ_EMPTY(&ctx->work))

    54

        // we can expect to get pthread_cancel'd here

    55

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

    56

    57

    // pop work

    58

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

    59

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

    60

    61

    // release

    62

    pthread_cleanup_pop(true);

    63

}

    64

    65

/**

    66

 * Wait for work queue to become empty

    67

 */

    68

static void pt_work_wait_idle (struct pt_ctx *ctx)

    69

{

    70

    // acquire

    71

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

    72

    73

    // wait for it to drain...

    74

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

    75

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

    76

    77

    // release

    78

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

    79

}

    80

    81

/**

    82

 * Execute a piece of work

    83

 */

    84

static void pt_work_execute (struct pt_work *work)

    85

{

    86

    work->func(work->arg);

    87

}

    88

    89

/**

    90

 * Release work state once done

    91

 */

    92

static void pt_work_release (struct pt_work *work)

    93

{

    94

    free(work);

    95

}

    96

    97

/**

    98

 * Worker thread entry point

    99

 */

   100

static void* pt_thread_main (void *arg)

   101

{

   102

    struct pt_thread *thread = arg;

   103

    struct pt_work *work;

   104

   105

    // if only life were so simple...

   106

    while (true) {

   107

        pt_work_dequeue(thread->ctx, &work);

   108

        pt_work_execute(work);

   109

        pt_work_release(work);

   110

    }

   111

   112

    return NULL;

   113

}

   114

   115

/**

   116

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

   117

 *

   118

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

   119

 */

   120

static void pt_thread_shutdown (struct pt_thread *thread)

   121

{

   122

    // signal it to stop at next cancel point (i.e. when waiting for work)

   123

    if (pthread_cancel(thread->tid))

   124

        perror("pthread_cancel");

   125

   126

    // reap

   127

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

   128

        perror("pthread_join");

   129

   130

    // mark

   131

    thread->tid = 0;

   132

}

   133

   134

/**

   135

 * Release pt_thread state, aborting thread if running.

   136

 *

   137

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

   138

 */

   139

static void pt_thread_destroy (struct pt_thread *thread)

   140

{

   141

    if (thread->tid) {

   142

        // detach

   143

        if (pthread_detach(thread->tid))

   144

            perror("pthread_detach");

   145

   146

        // abort thread

   147

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

   148

            perror("pthread_detach");

   149

   150

    }

   151

   152

    // remove from pool

   153

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

   154

   155

    free(thread);

   156

}

   157

   158

   159

/**

   160

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

   161

 */

   162

static int pt_ctx_add_thread (struct pt_ctx *ctx)

   163

{

   164

    struct pt_thread *thread;

   165

    int err;

   166

   167

    // alloc

   168

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

   169

        return -PT_ERR_MEM;

   170

   171

    // init

   172

    thread->ctx = ctx;

   173

   174

    // start thread, default attrs

   175

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

   176

        JUMP_SET_ERROR(err, PT_ERR_PTHREAD_CREATE);

   177

   178

    // add to pool

   179

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

   180

   181

    // ok

   182

    return 0;

   183

   184

error:

   185

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

   186

    free(thread);

   187

   188

    return err;

   189

}

   190

   191

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

   192

{

   193

    struct pt_ctx *ctx;

   194

    int err;

   195

   196

    // alloc

   197

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

   198

        return -PT_ERR_MEM;

   199

   200

    // init

   201

    TAILQ_INIT(&ctx->threads);

   202

    TAILQ_INIT(&ctx->work);

   203

    pthread_mutex_init(&ctx->work_mutex, NULL);

   204

    pthread_cond_init(&ctx->work_cond, NULL);

   205

    pthread_cond_init(&ctx->idle_cond, NULL);

   206

   207

    // start threadpool

   208

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

   209

        if ((err = pt_ctx_add_thread(ctx)))

   210

            JUMP_ERROR(err);

   211

    }

   212

   213

    ctx->running = true;

   214

   215

    // ok

   216

    *ctx_ptr = ctx;

   217

   218

    return 0;

   219

   220

error:

   221

    // cleanup

   222

    pt_ctx_destroy(ctx);

   223

   224

    return err;

   225

}

   226

   227

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

   228

{

   229

    struct pt_work *work;

   230

   231

    // check state

   233

    if (!ctx->running)

   234

        RETURN_ERROR(PT_ERR_CTX_SHUTDOWN);

   235

   237

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

   238

        RETURN_ERROR(PT_ERR_MEM);

   239

   240

    // init

   241

    work->func = func;

   242

    work->arg = arg;

   243

   244

    // enqueue

   245

    pt_work_enqueue(ctx, work);

   246

   247

    // ok

   248

    return 0;

   249

}

   250

   251

int pt_ctx_shutdown (struct pt_ctx *ctx)

   252

{

   253

    struct pt_thread *thread;

   254

   255

    // stop accepting new work

   256

    ctx->running = false;

   257

   258

    // wait for work queue to empty

   259

    pt_work_wait_idle(ctx);

   260

   261

    // shut down each thread

   262

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

   263

        // XXX: handle errors of some kind?

   264

        pt_thread_shutdown(thread);

   265

   266

    // then drop

   267

    pt_ctx_destroy(ctx);

   268

   269

    return 0;

   270

}

   271

   272

void pt_ctx_destroy (struct pt_ctx *ctx)

   273

{

   274

    struct pt_thread *thread;

   275

   276

    // kill threads

   277

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

   278

        pt_thread_destroy(thread);

   279

   280

    // destroy mutex/conds

   281

    if (pthread_cond_destroy(&ctx->idle_cond))

   282

        perror("pthread_cond_destroy(idle_cond)");

   283

   284

    if (pthread_cond_destroy(&ctx->work_cond))

   285

        perror("pthread_cond_destroy(work_cond)");

   286

   287

    if (pthread_mutex_destroy(&ctx->work_mutex))

   288

        perror("pthread_mutex_destroy(work_mutex)");

   289

   290

   291

    free(ctx);

   292

}

   293