1

#include <stdlib.h>

     2

#include <assert.h>

     3

     4

#include <pthread.h>

     5

     6

#include "common.h"

     7

#include "render_threads.h"

     8

#include "render_slices_struct.h"

     9

#include "render_mandelbrot.h"

    10

    11

/*

    12

 * Render a mandelbrot in several slices simultaneously using threads. 

    13

 *

    14

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

    15

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

    16

 * 

    17

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

    18

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

    19

 *  

    20

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

    21

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

    22

 *    processed thse segment in it

    23

 *      * process the segment

    24

 *

    25

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

    26

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

    27

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

    28

 *        the segment and this step,

    29

 *          * wait on the cond

    30

 *      * we can proceed to render the next segment

    31

 *          * we should now be marked as continueable

    32

 *

    33

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

    34

 *    newly completed row. 

    35

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

    36

 *        to process the row we just filled

    37

 *          * mark all the segments as continueable

    38

 *          * signal the cond the threads may be waiting on

    39

 *      * increment the waiting_rows counter

    40

 *          * signal the manager in case it's waiting

    41

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

    42

 *        segment,

    43

 *          * wait on the cond

    44

 *      * we can proceed to render the next segment

    45

 *          * we should now be marked as continueable

    46

 *

    47

 * * if we have no more segments to submit

    48

 *      * update the slices_running counter

    49

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

    50

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

    51

 *

    52

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

    53

 * have completed.

    54

 *

    55

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

    56

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

    57

 *   running)

    58

 *      * wait on a cond until something happens

    59

 *

    60

 * * if waiting_rows is nonzero

    61

 *      * decrement it

    62

 *      * process a row

    63

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

    64

 *          * mark all the segments as continueable

    65

 *          * signal the cond the threads may be waiting on

    66

 *

    67

 * * if slices_running is zero

    68

 *      * break out of the loop

    69

 *

    70

 * * otherwise, go back to the start again

    71

 */     

    72

    73

struct render_threads {

    74

    // do I need to free this?

    75

    int owned_by_me;

    76

    77

    // info for the induvidual threads

    78

    struct render_thread {

    79

        // the slice that we need to render

    80

        struct render_slice_info *slice_info;

    81

    82

        // our thread id

    83

        pthread_t thread;

    84

    85

        // how many times we are allowed to continue now

    86

        int can_continue;

    87

    88

        // how many segments we've written

    89

        int segments_done;

    90

    91

        // ptr back to ctx

    92

        struct render_threads *self;

    93

    94

    } threads[RENDER_SLICES_MAX];

    95

    96

    // the render_slices struct

    97

    struct render_slices slices_info;

    98

    99

    // the manager thread id

   100

    pthread_t manager_thread;

   101

   102

    // how many slices?

   103

    int slice_count;

   104

   105

    /*

   106

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

   107

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

   108

     *

   109

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

   110

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

   111

     *

   112

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

   113

     * continue_mut as normal.

   114

     */

   115

    int manager_running;

   116

   117

    /*

   118

     * Signals to the manager

   119

     */

   120

    int slices_running;

   121

    int waiting_rows;

   122

   123

    /* the inter-thread communication stuff */

   124

   125

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

   126

    // process_row now.

   127

    pthread_cond_t process_row_cond;

   128

    pthread_mutex_t process_row_mut;

   129

   130

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

   131

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

   132

    // on this cond.

   133

    pthread_cond_t continue_cond;

   134

    pthread_mutex_t continue_mut;

   135

   136

    // used to protect slices_info

   137

    pthread_mutex_t slices_mut;

   138

};

   139

   140

void render_threads_deinit (struct render_threads *ctx) {

   141

    render_slices_deinit(&ctx->slices_info);

   142

}

   143

   144

void render_threads_free (struct render_threads *ctx) {

   145

    render_threads_deinit(ctx);

   146

   147

    if (ctx->owned_by_me)

   148

        free(ctx);

   149

}

   150

   151

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

   152

    struct render_thread *subctx = arg;

   153

    int status, i;

   154

   155

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

   156

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

   157

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

   158

   159

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

   160

    subctx->can_continue = 0;

   161

   162

    // process the segment

   163

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

   164

   165

    // debugging

   166

/*    subctx->segments_done++;

   167

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

   168

        status & SLICE_PROCESS_ROW ? "SLICE_PROCESS_ROW" : "", 

   169

        status & SLICE_CONTINUE ? "SLICE_CONTINUE" : "",

   170

        subctx->segments_done

   171

    );

   172

*/

   173

   174

    // done with the slices stuff

   175

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

   176

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

   177

   178

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

   179

    if (status & SLICE_CONTINUE) {

   180

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

   181

   182

//        printf("slice %zu: signal continue\n", subctx->slice_info->index);

   183

   184

        // mark them as continueable

   185

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

   186

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

   187

   188

        // signal then to continue

   189

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

   190

   191

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

   192

    }

   193

   194

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

   195

    if (status & SLICE_PROCESS_ROW) {

   196

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

   197

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

   198

   199

        // tell it that it has a row waiting

   200

        subctx->self->waiting_rows++;

   201

   202

        // signal it in case it was waiting

   203

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

   204

   205

        // release the process_row mutex

   206

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

   207

    }

   208

   209

    // handle our can-continue status

   210

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

   211

   212

    if (status & SLICE_CONTINUE) {

   213

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

   214

//        printf("slice %zu: direct continue\n", subctx->slice_info->index);

   215

   216

    } else {

   217

        if (!subctx->can_continue) {

   218

            // we need to wait until someone signals it

   219

//            printf("slice %zu: waiting...\n", subctx->slice_info->index);

   220

   221

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

   222

   223

//            printf("slice %zu: continue after wait\n", subctx->slice_info->index);

   224

   225

        } else {

   226

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

   227

//            printf("slice %zu: indirect continue\n", subctx->slice_info->index);

   228

   229

        }

   230

    }

   231

   232

    // we should now be marked as continueable

   233

    assert(subctx->can_continue);

   234

   235

    // proceed to continue, so clear this

   236

    subctx->can_continue = 0;

   237

   238

    // done handling the continue state

   239

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

   240

   241

    // row handled succesfully

   242

    return 0;

   243

}

   244

   245

void *_render_threads_slice_func (void *arg) {

   246

    struct render_thread *subctx = arg;

   247

   248

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

   249

    // use _render_threads_slice_row to handle the completed rows

   250

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

   251

        &subctx->slice_info->render_buf, 

   252

        &_render_threads_slice_row, subctx)

   253

    )

   254

        goto error;

   255

   256

    // wait for the manager to start up

   257

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

   258

    if (!subctx->self->manager_running)

   259

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

   260

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

   261

   262

    // start rendering...

   263

//    printf("slice %zu: start\n", subctx->slice_info->index);

   264

   265

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

   266

        goto error;

   267

   268

    // success, slice render complete

   269

//    printf("slice %zu: done\n", subctx->slice_info->index);

   270

   271

    // sanity checks

   272

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

   273

   274

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

   275

   276

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

   277

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

   278

//        printf("slice %zu: signal\n", subctx->slice_info->index);

   279

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

   280

    }

   281

   282

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

   283

   284

    // successful exit

   285

    pthread_exit(NULL);

   286

   287

error:

   288

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

   289

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

   290

}

   291

   292

void *_render_threads_manager_func (void *arg) {

   293

    struct render_threads *ctx = arg;

   294

   295

    // the process_row return status

   296

    int status, i;

   297

   298

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

   299

    pthread_mutex_lock(&ctx->continue_mut);

   300

    ctx->manager_running = 1;

   301

    pthread_cond_broadcast(&ctx->continue_cond);

   302

    pthread_mutex_unlock(&ctx->continue_mut);

   303

   304

    // needs to be locked inside the loop

   305

    pthread_mutex_lock(&ctx->process_row_mut);

   306

   307

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

   308

    do {

   309

        // the process_row must be locked here

   310

//        printf("manager: to wait\n");

   311

   312

        // figure out if we need to wait

   313

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

   314

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

   315

//            printf("manager: waiting\n");

   316

   317

            // wait until a worker thread signals us

   318

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

   319

        }

   320

   321

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

   322

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

   323

   324

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

   325

   326

        // do we need to process a row?

   327

        if (ctx->waiting_rows > 0) {

   328

            // claim the row for ourself

   329

            ctx->waiting_rows--;

   330

   331

            // unlock the process_row mut while we handle this row

   332

            pthread_mutex_unlock(&ctx->process_row_mut);

   333

   334

   335

            // keep calls to render_slices synchronized

   336

            pthread_mutex_lock(&ctx->slices_mut);

   337

   338

            // call into render_slices

   339

            status = render_slices_process_row(&ctx->slices_info);

   340

   341

/*            printf("manager: did process_row, status=(%s %s)\n", 

   342

                status & SLICE_PROCESS_ROW ? "SLICE_PROCESS_ROW" : "", 

   343

                status & SLICE_CONTINUE ? "SLICE_CONTINUE" : ""

   344

            );

   345

*/            

   346

   347

            // done with render_slices

   348

            pthread_mutex_unlock(&ctx->slices_mut);

   349

   350

   351

            // any errors?

   352

            if (status == -1) {

   353

                goto error;

   354

            }

   355

   356

            // can any waiting slices now continue?

   357

            if (status & SLICE_CONTINUE) {

   358

                // grab the continue_mut lock while we update this state

   359

                pthread_mutex_lock(&ctx->continue_mut);

   360

   361

                // increment continue_count for all slices

   362

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

   363

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

   364

   365

//                printf("manager: signal continue\n");

   366

   367

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

   368

                pthread_cond_broadcast(&ctx->continue_cond);

   369

   370

                // done with that

   371

                pthread_mutex_unlock(&ctx->continue_mut);

   372

            }

   373

   374

            // XXX: ignore SLICE_PROCESS_ROW

   375

   376

            // grab our mutex again

   377

            pthread_mutex_lock(&ctx->process_row_mut);

   378

        }

   379

   380

        // are all the slices done now?

   381

        if (ctx->slices_running == 0) {

   382

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

   383

            pthread_mutex_unlock(&ctx->process_row_mut);

   384

   385

            break;

   386

        }

   387

   388

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

   389

    } while (1);

   390

   391

    // all slices are done now

   392

    if (render_slices_done(&ctx->slices_info))

   393

        goto error;

   394

   395

//    printf("manager: done\n");

   396

   397

    // reap all the workers

   398

    void *retval;

   399

   400

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

   401

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

   402

            PERROR("pthread_join");

   403

   404

        // XXX: assume they don't get canceled (SIGSEV etc?)

   405

        assert(retval == NULL);

   406

    }

   407

   408

    // ok, exit ourself

   409

    pthread_exit(NULL);

   410

   411

error:

   412

    /* XXX: handle errors gracefully... */

   413

    FATAL("the threaded render operation failed, taking down the rest of the daemon application as well...");

   414

}

   415

   416

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

   417

    // break the render operation down into induvidual slices

   418

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

   419

        goto error;

   420

   421

    // init the mutexes (this should never fail)

   422

    assert(!(

   423

            pthread_mutex_init(&ctx->process_row_mut, NULL)

   424

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

   425

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

   426

    ));

   427

   428

    // init the conds (this should never fail)

   429

    assert(!(

   430

            pthread_cond_init(&ctx->process_row_cond, NULL)

   431

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

   432

    ));

   433

   434

    // how many slices?

   435

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

   436

   437

    // spawn a thread for each slice

   438

    int index;

   439

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

   440

        // the self ptr...

   441

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

   442

   443

        // fetch the slice info

   444

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

   445

   446

        // spawn the thread

   447

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

   448

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

   449

    }

   450

   451

    // init the attrs for the manager thread

   452

    pthread_attr_t manager_attrs;

   453

   454

    if (

   455

            pthread_attr_init(&manager_attrs)

   456

//         || pthread_setdetachstate(&manager_attrs, PTHREAD_CREATE_DETACHED)

   457

    )

   458

        PERROR("pthread_attr_init/pthread_attr_setdetachstate");

   459

   460

    // spawn the manager thread

   461

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

   462

        PERROR("pthread_create(manager_func)");

   463

   464

    // success

   465

    return 0;

   466

   467

error:

   468

    render_threads_deinit(ctx);

   469

    return -1;

   470

}

   471

   472

struct render_threads *render_threads_alloc (struct render *render_info) {

   473

    struct render_threads *ctx = NULL;

   474

   475

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

   476

        ERROR("calloc");

   477

   478

    // init with silent fall-through

   479

    if (render_threads_init(ctx, render_info))

   480

        goto error;

   481

   482

    // success

   483

    return ctx;

   484

   485

error:

   486

    return NULL;

   487

}

   488

   489

int render_threads_wait (struct render_threads *ctx) {

   490

    void *retval;

   491

   492

    if (pthread_join(ctx->manager_thread, &retval))

   493

        PERROR("pthread_join");

   494

   495

    if (retval == PTHREAD_CANCELED)

   496

        ERROR("manager thread was canceled");

   497

    else if (retval != NULL)

   498

        ERROR("manager thread exited with unknown return value");

   499

   500

    // ok, success

   501

    return 0;

   502

   503

error:

   504

    // caller needs to free ctx

   505

    return -1;

   506

}

   507