1

#include <stdlib.h>

     2

#include <assert.h>

     3

     4

#include <pthread.h>

     5

     6

//#define DEBUG_ENABLED

     7

     8

#include "common.h"

     9

#include "render_threads.h"

    10

#include "render_slices_struct.h"

    11

#include "render_mandelbrot.h"

    12

    13

/*

    14

 * Render a mandelbrot in several slices simultaneously using threads. 

    15

 *

    16

 * This works as an n:1 producer-consumer model, where we have n worker threads

    17

 * rendering n slices, and one manager thread that processes the completed rows.

    18

 * 

    19

 * The worker threads function by first rendering a row of pixel data into memory,

    20

 * and then calling a function that will do something with that completed segment.

    21

 *  

    22

 *  * clear the can_continue marker, because the real value may change depending

    23

 *    on what our segment does, and we don't want stale values from before we

    24

 *    processed thse segment in it

    25

 *      * process the segment

    26

 *

    27

 *  * if the segment didn't complete a row, wait until some worker thread submits

    28

 *    a segment that does, or the manager thread processes a row

    29

 *      * if noone has signaled the cond in between us deciding what to do with

    30

 *        the segment and this step,

    31

 *          * wait on the cond

    32

 *      * we can proceed to render the next segment

    33

 *          * we should now be marked as continueable

    34

 *

    35

 *  * if the segment completed a row, the manager thread needs to process this

    36

 *    newly completed row. 

    37

 *      * if we can also continue on to the next row without waiting for the manager

    38

 *        to process the row we just filled

    39

 *          * mark all the segments as continueable

    40

 *          * signal the cond the threads may be waiting on

    41

 *      * increment the waiting_rows counter

    42

 *          * signal the manager in case it's waiting

    43

 *      * else, if noone has signaled the continue cond after we processed the

    44

 *        segment,

    45

 *          * wait on the cond

    46

 *      * we can proceed to render the next segment

    47

 *          * we should now be marked as continueable

    48

 *

    49

 * * if we have no more segments to submit

    50

 *      * update the slices_running counter

    51

 *          * if the counter is now zero (we were the last slice to complete)

    52

 *              * signal the manager thread that it doesn't need to wait anymore

    53

 *

    54

 * The manager thread just sits in a loop calling process_row until all the slices

    55

 * have completed.

    56

 *

    57

 * * if no workers have updated the waiting_rows counter since we last processed

    58

 *   a row, and the slices_running counter isn't zero (we still have threads

    59

 *   running)

    60

 *      * wait on a cond until something happens

    61

 *

    62

 * * if waiting_rows is nonzero

    63

 *      * decrement it

    64

 *      * process a row

    65

 *      * if processing the row means that waiting slices can continue

    66

 *          * mark all the segments as continueable

    67

 *          * signal the cond the threads may be waiting on

    68

 *

    69

 * * if slices_running is zero

    70

 *      * break out of the loop

    71

 *

    72

 * * otherwise, go back to the start again

    73

 */     

    74

    75

struct render_threads {

    76

    // do I need to free this?

    77

    int owned_by_me;

    78

    79

    // info for the induvidual threads

    80

    struct render_thread {

    81

        // the slice that we need to render

    82

        struct render_slice_info *slice_info;

    83

    84

        // our thread id

    85

        pthread_t thread;

    86

    87

        // how many times we are allowed to continue now

    88

        int can_continue;

    89

    90

        // how many segments we've written

    91

        int segments_done;

    92

    93

        // ptr back to ctx

    94

        struct render_threads *self;

    95

    96

    } threads[RENDER_SLICES_MAX];

    97

    98

    // the render_slices struct

    99

    struct render_slices slices_info;

   100

   101

    // the manager thread id

   102

    pthread_t manager_thread;

   103

   104

    // how many slices?

   105

    int slice_count;

   106

   107

    /*

   108

     * The worker threads must be careful to wait for the manager thread to start running before they attempt to signal

   109

     * the process_row_cond. This is done via this flag, in conjunction with continue_{cond,mut}. 

   110

     *

   111

     * Workers should lock continue_mut, and if this flag is zero, wait on continue_cond (releasing the mutex). If the

   112

     * flag is already nonzero, the worker threads do not need to cond_wait and can just continue.

   113

     *

   114

     * On startup, the manager thread will lock continue_mut, set this to nonzero, signal continue_cond, and unlock

   115

     * continue_mut as normal.

   116

     */

   117

    int manager_running;

   118

   119

    /*

   120

     * Signals to the manager

   121

     */

   122

    int slices_running;

   123

    int waiting_rows;

   124

   125

    /* the inter-thread communication stuff */

   126

   127

    // the worker threads signal this when their segment_done call indicates that the manager thread should call

   128

    // process_row now.

   129

    pthread_cond_t process_row_cond;

   130

    pthread_mutex_t process_row_mut;

   131

   132

    // the worker threads and the manager thread signal this when segment_done or process_row indicate that the worker

   133

    // threads should call segment_done again. Worker threads that receive ~SLICE_CONTINUE from segment_done will wait

   134

    // on this cond.

   135

    pthread_cond_t continue_cond;

   136

    pthread_mutex_t continue_mut;

   137

   138

    // used to protect slices_info

   139

    pthread_mutex_t slices_mut;

   140

   141

    /*

   142

     * Return value from the manager thread

   143

     */

   144

    enum render_threads_retval {

   145

        RETVAL_SUCCESS,

   146

        RETVAL_FAILURE,

   147

    } manager_retval;

   148

};

   149

   150

   151

void render_threads_deinit (struct render_threads *ctx) {

   152

    render_slices_deinit(&ctx->slices_info);

   153

}

   154

   155

void render_threads_free (struct render_threads *ctx) {

   156

    render_threads_deinit(ctx);

   157

   158

    if (ctx->owned_by_me)

   159

        free(ctx);

   160

}

   161

   162

int _render_threads_slice_row (void *arg, unsigned char *rowbuf) {

   163

    struct render_thread *subctx = arg;

   164

    int status, i;

   165

   166

    // test for cancelation, we don't hold any locks right now

   167

    pthread_testcancel();

   168

   169

    // we need to handle the render_slices state atomically, so lock it

   170

    pthread_mutex_lock(&subctx->self->slices_mut);

   171

    pthread_mutex_lock(&subctx->self->continue_mut);

   172

   173

    // make sure that we don't have any stale state in this

   174

    subctx->can_continue = 0;

   175

   176

    // process the segment

   177

    status = render_slices_segment_done(&subctx->self->slices_info, subctx->slice_info->index);

   178

   179

    // debugging

   180

    DEBUG("slice %zu: segment_done, status=(%s %s), row=%d", subctx->slice_info->index,

   181

        status & SLICE_PROCESS_ROW ? "SLICE_PROCESS_ROW" : "", 

   182

        status & SLICE_CONTINUE ? "SLICE_CONTINUE" : "",

   183

        ++subctx->segments_done

   184

    );

   185

   186

    // do we need to notify the other worker threads?

   187

    if (status & SLICE_CONTINUE) {

   188

        DEBUG("slice %zu: signal continue", subctx->slice_info->index);

   189

   190

        // mark them as continueable

   191

        for (i = 0; i < subctx->self->slice_count; i++)

   192

            subctx->self->threads[i].can_continue = 1;

   193

   194

        // signal then to continue

   195

        pthread_cond_broadcast(&subctx->self->continue_cond);

   196

    }

   197

   198

    // done with the slices stuff

   199

    pthread_mutex_unlock(&subctx->self->continue_mut);

   200

    pthread_mutex_unlock(&subctx->self->slices_mut);

   201

   202

    // tell the manager thread that it can process a row

   203

    if (status & SLICE_PROCESS_ROW) {

   204

        // grab the process_row lock so that we can do stuff with it

   205

        pthread_mutex_lock(&subctx->self->process_row_mut);

   206

   207

        // tell it that it has a row waiting

   208

        subctx->self->waiting_rows++;

   209

   210

        // signal it in case it was waiting

   211

        pthread_cond_signal(&subctx->self->process_row_cond);

   212

   213

        // release the process_row mutex

   214

        pthread_mutex_unlock(&subctx->self->process_row_mut);

   215

    }

   216

   217

#define BEGIN_MUTEX(mut) pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock, &(mut)); pthread_mutex_lock(&(mut));

   218

#define END_MUTEX pthread_cleanup_pop(1);

   219

   220

    // handle our can-continue status

   221

    // we need to release this lock if we get canceled in pthread_cond_wait

   222

    BEGIN_MUTEX(subctx->self->continue_mut)

   223

        if (status & SLICE_CONTINUE) {

   224

            // don't wait for anything, we can just continue right away

   225

            DEBUG("slice %zu: direct continue", subctx->slice_info->index);

   226

   227

        } else {

   228

            if (!subctx->can_continue) {

   229

                // we need to wait until someone signals it

   230

                DEBUG("slice %zu: waiting...", subctx->slice_info->index);

   231

   232

                pthread_cond_wait(&subctx->self->continue_cond, &subctx->self->continue_mut);

   233

   234

                DEBUG("slice %zu: continue after wait", subctx->slice_info->index);

   235

   236

            } else {

   237

                // no need to wait, because someone else took care of that before we got a chance to wait

   238

                DEBUG("slice %zu: indirect continue", subctx->slice_info->index);

   239

   240

            }

   241

        }

   242

   243

        // we should now be marked as continueable

   244

        assert(subctx->can_continue);

   245

   246

        // proceed to continue, so clear this

   247

        subctx->can_continue = 0;

   248

   249

        // done handling the continue state

   250

        pthread_mutex_unlock(&subctx->self->continue_mut);

   251

   252

    END_MUTEX

   253

   254

    // row handled succesfully

   255

    return 0;

   256

}

   257

   258

void *_render_threads_slice_func (void *arg) {

   259

    struct render_thread *subctx = arg;

   260

   261

    // set up the render_info to render into local memory using the address provided via slice_info

   262

    // use _render_threads_slice_row to handle the completed rows

   263

    if (render_local_mem(subctx->slice_info->render_info, 

   264

        &subctx->slice_info->render_buf, 

   265

        &_render_threads_slice_row, subctx)

   266

    )

   267

        goto error;

   268

   269

    // wait for the manager to start up

   270

    pthread_mutex_lock(&subctx->self->continue_mut);

   271

    if (!subctx->self->manager_running)

   272

        pthread_cond_wait(&subctx->self->continue_cond, &subctx->self->continue_mut);

   273

    pthread_mutex_unlock(&subctx->self->continue_mut);

   274

   275

    // start rendering...

   276

    DEBUG("slice %zu: start", subctx->slice_info->index);

   277

   278

    if (render_mandelbrot(subctx->slice_info->render_info))

   279

        goto error;

   280

   281

    // success, slice render complete

   282

    DEBUG("slice %zu: done", subctx->slice_info->index);

   283

   284

    // sanity checks

   285

    assert(subctx->self->slices_running > 0);

   286

   287

    pthread_mutex_lock(&subctx->self->process_row_mut);

   288

   289

    if (--subctx->self->slices_running == 0) {

   290

        // this was the last row, send the manager a final signal in case it's waiting

   291

        DEBUG("slice %zu: signal", subctx->slice_info->index);

   292

        pthread_cond_signal(&subctx->self->process_row_cond);

   293

    }

   294

   295

    pthread_mutex_unlock(&subctx->self->process_row_mut);

   296

   297

    // successful exit

   298

    pthread_exit(NULL);

   299

   300

error:

   301

    /* XXX: signal an error condition? Mind, with the current code, this should never happen... */

   302

    FATAL("a render thread failed somehow, taking down the rest of the daemon application as well");

   303

}

   304

   305

void *_render_threads_manager_func (void *arg) {

   306

    struct render_threads *ctx = arg;

   307

   308

    // the process_row return status

   309

    int status, i;

   310

   311

    // first, mark us as running and signal any waiting workers

   312

    pthread_mutex_lock(&ctx->continue_mut);

   313

    ctx->manager_running = 1;

   314

    pthread_cond_broadcast(&ctx->continue_cond);

   315

    pthread_mutex_unlock(&ctx->continue_mut);

   316

   317

    // needs to be locked inside the loop

   318

    pthread_mutex_lock(&ctx->process_row_mut);

   319

   320

    // then we wait around and call process_row when told to

   321

    do {

   322

        // the process_row must be locked here

   323

        DEBUG("manager: to wait");

   324

   325

        // figure out if we need to wait

   326

        if (ctx->waiting_rows == 0 && ctx->slices_running > 0) {

   327

            // no work to do right now, so wait for some

   328

            DEBUG("manager: waiting");

   329

   330

            // wait until a worker thread signals us

   331

            pthread_cond_wait(&ctx->process_row_cond, &ctx->process_row_mut);

   332

        }

   333

   334

        // make sure we actually have something to do now...

   335

        assert(ctx->waiting_rows > 0 || ctx->slices_running == 0);

   336

   337

        // keep the process_row mutex locked until we've inspected/updated both vars

   338

   339

        // do we need to process a row?

   340

        if (ctx->waiting_rows > 0) {

   341

            // claim the row for ourself

   342

            ctx->waiting_rows--;

   343

   344

            // unlock the process_row mut while we handle this row

   345

            pthread_mutex_unlock(&ctx->process_row_mut);

   346

   347

   348

            // keep calls to render_slices synchronized

   349

            pthread_mutex_lock(&ctx->slices_mut);

   350

   351

            // call into render_slices

   352

            status = render_slices_process_row(&ctx->slices_info);

   353

   354

            DEBUG("manager: did process_row, status=(%s %s)", 

   355

                status & SLICE_PROCESS_ROW ? "SLICE_PROCESS_ROW" : "", 

   356

                status & SLICE_CONTINUE ? "SLICE_CONTINUE" : ""

   357

            );

   358

   359

            // any errors?

   360

            if (status == -1) {

   361

                pthread_mutex_unlock(&ctx->slices_mut);

   362

                goto error;

   363

            }

   364

   365

            // can any waiting slices now continue?

   366

            if (status & SLICE_CONTINUE) {

   367

                // grab the continue_mut lock while we update this state

   368

                pthread_mutex_lock(&ctx->continue_mut);

   369

   370

                // set can_continue for all slices

   371

                for (i = 0; i < ctx->slice_count; i++)

   372

                    ctx->threads[i].can_continue = 1;

   373

   374

                DEBUG("manager: signal continue");

   375

   376

                // signal any waiting worker threads to continue, giving them a chance to call cond_wait first.

   377

                pthread_cond_broadcast(&ctx->continue_cond);

   378

   379

                // done with that

   380

                pthread_mutex_unlock(&ctx->continue_mut);

   381

            }

   382

   383

            // done with render_slices

   384

            pthread_mutex_unlock(&ctx->slices_mut);

   385

   386

            // XXX: ignore SLICE_PROCESS_ROW

   387

   388

            // grab our mutex again

   389

            pthread_mutex_lock(&ctx->process_row_mut);

   390

        }

   391

   392

        // are all the rows and slices done now?

   393

        if (ctx->slices_running == 0 && ctx->waiting_rows == 0) {

   394

            // unlock the mutex so the worker threads don't get stuck on it

   395

            pthread_mutex_unlock(&ctx->process_row_mut);

   396

   397

            break;

   398

        }

   399

   400

        // keep the process_row mutex locked at the top of the loop 

   401

    } while (1);

   402

   403

    // all slices are done now

   404

    if (render_slices_done(&ctx->slices_info))

   405

        goto error;

   406

   407

    DEBUG("manager: done");

   408

   409

    void *retval;

   410

   411

    // reap all the workers

   412

    for (i = 0; i < ctx->slice_count; i++) {

   413

        if (pthread_join(ctx->threads[i].thread, &retval))

   414

            PERROR("pthread_join");

   415

   416

        // XXX: assume they don't get canceled

   417

        assert(retval == NULL);

   418

    }

   419

   420

    // succesfull return

   421

    ctx->manager_retval = RETVAL_SUCCESS;

   422

   423

    goto exit;

   424

   425

error:

   426

    /* cancel and join all worker threads */

   427

    WARNING("canceling and reaping all threads");

   428

   429

    for (i = 0; i < ctx->slice_count; i++) {

   430

        if (pthread_cancel(ctx->threads[i].thread))

   431

            PWARNING("pthread_cancel(worker:%d)", i);

   432

   433

        // join it in any case

   434

        if (pthread_join(ctx->threads[i].thread, &retval)) {

   435

            PWARNING("pthread_join(worker:%d)", i);

   436

        } else {

   437

            if (retval != PTHREAD_CANCELED)

   438

                PWARNING("worker:%d retval = %p", i, retval);

   439

        }

   440

    }

   441

   442

    // failure

   443

    ctx->manager_retval = RETVAL_FAILURE;

   444

   445

exit:    

   446

    // ok, exit ourself

   447

    pthread_exit(ctx);

   448

}

   449

   450

int render_threads_init (struct render_threads *ctx, struct render *render_info) {

   451

    // break the render operation down into induvidual slices

   452

    if (render_slices_init(&ctx->slices_info, render_info))

   453

        goto error;

   454

   455

    // init the mutexes (this should never fail)

   456

    assert(!(

   457

            pthread_mutex_init(&ctx->process_row_mut, NULL)

   458

         || pthread_mutex_init(&ctx->continue_mut, NULL)

   459

         || pthread_mutex_init(&ctx->slices_mut, NULL)

   460

    ));

   461

   462

    // init the conds (this should never fail)

   463

    assert(!(

   464

            pthread_cond_init(&ctx->process_row_cond, NULL)

   465

         || pthread_cond_init(&ctx->continue_cond, NULL)

   466

    ));

   467

   468

    // how many slices?

   469

    ctx->slices_running = ctx->slice_count = render_slices_get_count(&ctx->slices_info);

   470

   471

    // spawn a thread for each slice

   472

    int index;

   473

    for (index = 0; index < ctx->slice_count; index++) {

   474

        // the self ptr...

   475

        ctx->threads[index].self = ctx;

   476

   477

        // fetch the slice info

   478

        ctx->threads[index].slice_info = render_slices_get_slice_info(&ctx->slices_info, index);

   479

   480

        // spawn the thread

   481

        if (pthread_create(&ctx->threads[index].thread, NULL, &_render_threads_slice_func, &ctx->threads[index]))

   482

            PERROR("pthread_create(slice_func:%d)", index);

   483

    }

   484

   485

    // init the attrs for the manager thread

   486

    pthread_attr_t manager_attrs;

   487

   488

    if (

   489

            pthread_attr_init(&manager_attrs)

   490

//         || pthread_setdetachstate(&manager_attrs, PTHREAD_CREATE_DETACHED)

   491

    )

   492

        PERROR("pthread_attr_init/pthread_attr_setdetachstate");

   493

   494

    // spawn the manager thread

   495

    if (pthread_create(&ctx->manager_thread, &manager_attrs, &_render_threads_manager_func, ctx))

   496

        PERROR("pthread_create(manager_func)");

   497

   498

    // success

   499

    return 0;

   500

   501

error:

   502

    render_threads_deinit(ctx);

   503

    return -1;

   504

}

   505

   506

struct render_threads *render_threads_alloc (struct render *render_info) {

   507

    struct render_threads *ctx = NULL;

   508

   509

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

   510

        ERROR("calloc");

   511

   512

    // flag it for free()ing

   513

    ctx->owned_by_me = 1;

   514

   515

    // init with silent fall-through

   516

    if (render_threads_init(ctx, render_info))

   517

        goto error;

   518

   519