granicus.if.org Git - libvpx/blob - vp9/encoder/vp9_multi_thread.c

   1 /*
   2  *  Copyright (c) 2017 The WebM project authors. All Rights Reserved.
   3  *
   4  *  Use of this source code is governed by a BSD-style license
   5  *  that can be found in the LICENSE file in the root of the source
   6  *  tree. An additional intellectual property rights grant can be found
   7  *  in the file PATENTS.  All contributing project authors may
   8  *  be found in the AUTHORS file in the root of the source tree.
   9  */
  10
  11 #include <assert.h>
  12
  13 #include "vp9/encoder/vp9_encoder.h"
  14 #include "vp9/encoder/vp9_ethread.h"
  15 #include "vp9/encoder/vp9_multi_thread.h"
  16
  17 void *vp9_enc_grp_get_next_job(MultiThreadHandle *multi_thread_ctxt,
  18                                int tile_id) {
  19   RowMTInfo *row_mt_info;
  20   JobQueueHandle *job_queue_hdl = NULL;
  21   void *next = NULL;
  22   JobNode *job_info = NULL;
  23 #if CONFIG_MULTITHREAD
  24   pthread_mutex_t *mutex_handle = NULL;
  25 #endif
  26
  27   row_mt_info = (RowMTInfo *)(&multi_thread_ctxt->row_mt_info[tile_id]);
  28   job_queue_hdl = (JobQueueHandle *)&row_mt_info->job_queue_hdl;
  29 #if CONFIG_MULTITHREAD
  30   mutex_handle = &row_mt_info->job_mutex;
  31 #endif
  32
  33 // lock the mutex for queue access
  34 #if CONFIG_MULTITHREAD
  35   pthread_mutex_lock(mutex_handle);
  36 #endif
  37   next = job_queue_hdl->next;
  38   if (NULL != next) {
  39     JobQueue *job_queue = (JobQueue *)next;
  40     job_info = &job_queue->job_info;
  41     // Update the next job in the queue
  42     job_queue_hdl->next = job_queue->next;
  43     job_queue_hdl->num_jobs_acquired++;
  44   }
  45
  46 #if CONFIG_MULTITHREAD
  47   pthread_mutex_unlock(mutex_handle);
  48 #endif
  49
  50   return job_info;
  51 }
  52
  53 void vp9_row_mt_mem_alloc(VP9_COMP *cpi) {
  54   struct VP9Common *cm = &cpi->common;
  55   MultiThreadHandle *multi_thread_ctxt = &cpi->multi_thread_ctxt;
  56   int tile_row, tile_col;
  57   const int tile_cols = 1 << cm->log2_tile_cols;
  58   const int tile_rows = 1 << cm->log2_tile_rows;
  59   const int sb_rows = mi_cols_aligned_to_sb(cm->mi_rows) >> MI_BLOCK_SIZE_LOG2;
  60   int jobs_per_tile_col, total_jobs;
  61
  62   jobs_per_tile_col = VPXMAX(cm->mb_rows, sb_rows);
  63   // Calculate the total number of jobs
  64   total_jobs = jobs_per_tile_col * tile_cols;
  65
  66   multi_thread_ctxt->allocated_tile_cols = tile_cols;
  67   multi_thread_ctxt->allocated_tile_rows = tile_rows;
  68   multi_thread_ctxt->allocated_vert_unit_rows = jobs_per_tile_col;
  69
  70   multi_thread_ctxt->job_queue =
  71       (JobQueue *)vpx_memalign(32, total_jobs * sizeof(JobQueue));
  72
  73 #if CONFIG_MULTITHREAD
  74   // Create mutex for each tile
  75   for (tile_col = 0; tile_col < tile_cols; tile_col++) {
  76     RowMTInfo *row_mt_info = &multi_thread_ctxt->row_mt_info[tile_col];
  77     pthread_mutex_init(&row_mt_info->job_mutex, NULL);
  78   }
  79 #endif
  80
  81   // Allocate memory for row based multi-threading
  82   for (tile_col = 0; tile_col < tile_cols; tile_col++) {
  83     TileDataEnc *this_tile = &cpi->tile_data[tile_col];
  84     vp9_row_mt_sync_mem_alloc(&this_tile->row_mt_sync, cm, jobs_per_tile_col);
  85     if (cpi->sf.adaptive_rd_thresh_row_mt) {
  86       const int sb_rows =
  87           (mi_cols_aligned_to_sb(cm->mi_rows) >> MI_BLOCK_SIZE_LOG2) + 1;
  88       int i;
  89       this_tile->row_base_thresh_freq_fact =
  90           (int *)vpx_calloc(sb_rows * BLOCK_SIZES * MAX_MODES,
  91                             sizeof(*(this_tile->row_base_thresh_freq_fact)));
  92       for (i = 0; i < sb_rows * BLOCK_SIZES * MAX_MODES; i++)
  93         this_tile->row_base_thresh_freq_fact[i] = RD_THRESH_INIT_FACT;
  94     }
  95   }
  96
  97   // Assign the sync pointer of tile row zero for every tile row > 0
  98   for (tile_row = 1; tile_row < tile_rows; tile_row++) {
  99     for (tile_col = 0; tile_col < tile_cols; tile_col++) {
 100       TileDataEnc *this_tile = &cpi->tile_data[tile_row * tile_cols + tile_col];
 101       TileDataEnc *this_col_tile = &cpi->tile_data[tile_col];
 102       this_tile->row_mt_sync = this_col_tile->row_mt_sync;
 103     }
 104   }
 105
 106   // Calculate the number of vertical units in the given tile row
 107   for (tile_row = 0; tile_row < tile_rows; tile_row++) {
 108     TileDataEnc *this_tile = &cpi->tile_data[tile_row * tile_cols];
 109     TileInfo *tile_info = &this_tile->tile_info;
 110     multi_thread_ctxt->num_tile_vert_sbs[tile_row] =
 111         get_num_vert_units(*tile_info, MI_BLOCK_SIZE_LOG2);
 112   }
 113
 114 #if CONFIG_MULTITHREAD
 115   for (tile_row = 0; tile_row < tile_rows; tile_row++) {
 116     for (tile_col = 0; tile_col < tile_cols; tile_col++) {
 117       TileDataEnc *this_tile = &cpi->tile_data[tile_row * tile_cols + tile_col];
 118
 119       CHECK_MEM_ERROR(cm, this_tile->enc_row_mt_mutex,
 120                       vpx_malloc(sizeof(*this_tile->enc_row_mt_mutex)));
 121
 122       pthread_mutex_init(this_tile->enc_row_mt_mutex, NULL);
 123     }
 124   }
 125 #endif
 126 }
 127
 128 void vp9_row_mt_mem_dealloc(VP9_COMP *cpi) {
 129   MultiThreadHandle *multi_thread_ctxt = &cpi->multi_thread_ctxt;
 130   int tile_col;
 131 #if CONFIG_MULTITHREAD
 132   int tile_row;
 133 #endif
 134
 135   // Deallocate memory for job queue
 136   if (multi_thread_ctxt->job_queue) vpx_free(multi_thread_ctxt->job_queue);
 137
 138 #if CONFIG_MULTITHREAD
 139   // Destroy mutex for each tile
 140   for (tile_col = 0; tile_col < multi_thread_ctxt->allocated_tile_cols;
 141        tile_col++) {
 142     RowMTInfo *row_mt_info = &multi_thread_ctxt->row_mt_info[tile_col];
 143     if (row_mt_info) pthread_mutex_destroy(&row_mt_info->job_mutex);
 144   }
 145 #endif
 146
 147   // Free row based multi-threading sync memory
 148   for (tile_col = 0; tile_col < multi_thread_ctxt->allocated_tile_cols;
 149        tile_col++) {
 150     TileDataEnc *this_tile = &cpi->tile_data[tile_col];
 151     vp9_row_mt_sync_mem_dealloc(&this_tile->row_mt_sync);
 152   }
 153
 154 #if CONFIG_MULTITHREAD
 155   for (tile_row = 0; tile_row < multi_thread_ctxt->allocated_tile_rows;
 156        tile_row++) {
 157     for (tile_col = 0; tile_col < multi_thread_ctxt->allocated_tile_cols;
 158          tile_col++) {
 159       TileDataEnc *this_tile =
 160           &cpi->tile_data[tile_row * multi_thread_ctxt->allocated_tile_cols +
 161                           tile_col];
 162       if (cpi->sf.adaptive_rd_thresh_row_mt) {
 163         if (this_tile->row_base_thresh_freq_fact != NULL) {
 164           vpx_free(this_tile->row_base_thresh_freq_fact);
 165           this_tile->row_base_thresh_freq_fact = NULL;
 166         }
 167       }
 168       pthread_mutex_destroy(this_tile->enc_row_mt_mutex);
 169       vpx_free(this_tile->enc_row_mt_mutex);
 170       this_tile->enc_row_mt_mutex = NULL;
 171     }
 172   }
 173 #endif
 174 }
 175
 176 void vp9_multi_thread_tile_init(VP9_COMP *cpi) {
 177   VP9_COMMON *const cm = &cpi->common;
 178   const int tile_cols = 1 << cm->log2_tile_cols;
 179   const int sb_rows = mi_cols_aligned_to_sb(cm->mi_rows) >> MI_BLOCK_SIZE_LOG2;
 180   int i;
 181
 182   for (i = 0; i < tile_cols; i++) {
 183     TileDataEnc *this_tile = &cpi->tile_data[i];
 184     int jobs_per_tile_col = cpi->oxcf.pass == 1 ? cm->mb_rows : sb_rows;
 185
 186     // Initialize cur_col to -1 for all rows.
 187     memset(this_tile->row_mt_sync.cur_col, -1,
 188            sizeof(*this_tile->row_mt_sync.cur_col) * jobs_per_tile_col);
 189     vp9_zero(this_tile->fp_data);
 190     this_tile->fp_data.image_data_start_row = INVALID_ROW;
 191   }
 192 }
 193
 194 void vp9_assign_tile_to_thread(MultiThreadHandle *multi_thread_ctxt,
 195                                int tile_cols, int num_workers) {
 196   int tile_id = 0;
 197   int i;
 198
 199   // Allocating the threads for the tiles
 200   for (i = 0; i < num_workers; i++) {
 201     multi_thread_ctxt->thread_id_to_tile_id[i] = tile_id++;
 202     if (tile_id == tile_cols) tile_id = 0;
 203   }
 204 }
 205
 206 int vp9_get_job_queue_status(MultiThreadHandle *multi_thread_ctxt,
 207                              int cur_tile_id) {
 208   RowMTInfo *row_mt_info;
 209   JobQueueHandle *job_queue_hndl;
 210 #if CONFIG_MULTITHREAD
 211   pthread_mutex_t *mutex;
 212 #endif
 213   int num_jobs_remaining;
 214
 215   row_mt_info = &multi_thread_ctxt->row_mt_info[cur_tile_id];
 216   job_queue_hndl = &row_mt_info->job_queue_hdl;
 217 #if CONFIG_MULTITHREAD
 218   mutex = &row_mt_info->job_mutex;
 219 #endif
 220
 221 #if CONFIG_MULTITHREAD
 222   pthread_mutex_lock(mutex);
 223 #endif
 224   num_jobs_remaining =
 225       multi_thread_ctxt->jobs_per_tile_col - job_queue_hndl->num_jobs_acquired;
 226 #if CONFIG_MULTITHREAD
 227   pthread_mutex_unlock(mutex);
 228 #endif
 229
 230   return (num_jobs_remaining);
 231 }
 232
 233 void vp9_prepare_job_queue(VP9_COMP *cpi, JOB_TYPE job_type) {
 234   VP9_COMMON *const cm = &cpi->common;
 235   MultiThreadHandle *multi_thread_ctxt = &cpi->multi_thread_ctxt;
 236   JobQueue *job_queue = multi_thread_ctxt->job_queue;
 237   const int tile_cols = 1 << cm->log2_tile_cols;
 238   int job_row_num, jobs_per_tile, jobs_per_tile_col, total_jobs;
 239   const int sb_rows = mi_cols_aligned_to_sb(cm->mi_rows) >> MI_BLOCK_SIZE_LOG2;
 240   int tile_col, i;
 241
 242   jobs_per_tile_col = (job_type != ENCODE_JOB) ? cm->mb_rows : sb_rows;
 243   total_jobs = jobs_per_tile_col * tile_cols;
 244
 245   multi_thread_ctxt->jobs_per_tile_col = jobs_per_tile_col;
 246   // memset the entire job queue buffer to zero
 247   memset(job_queue, 0, total_jobs * sizeof(JobQueue));
 248
 249   // Job queue preparation
 250   for (tile_col = 0; tile_col < tile_cols; tile_col++) {
 251     RowMTInfo *tile_ctxt = &multi_thread_ctxt->row_mt_info[tile_col];
 252     JobQueue *job_queue_curr, *job_queue_temp;
 253     int tile_row = 0;
 254
 255     tile_ctxt->job_queue_hdl.next = (void *)job_queue;
 256     tile_ctxt->job_queue_hdl.num_jobs_acquired = 0;
 257
 258     job_queue_curr = job_queue;
 259     job_queue_temp = job_queue;
 260
 261     // loop over all the vertical rows
 262     for (job_row_num = 0, jobs_per_tile = 0; job_row_num < jobs_per_tile_col;
 263          job_row_num++, jobs_per_tile++) {
 264       job_queue_curr->job_info.vert_unit_row_num = job_row_num;
 265       job_queue_curr->job_info.tile_col_id = tile_col;
 266       job_queue_curr->job_info.tile_row_id = tile_row;
 267       job_queue_curr->next = (void *)(job_queue_temp + 1);
 268       job_queue_curr = ++job_queue_temp;
 269
 270       if (ENCODE_JOB == job_type) {
 271         if (jobs_per_tile >=
 272             multi_thread_ctxt->num_tile_vert_sbs[tile_row] - 1) {
 273           tile_row++;
 274           jobs_per_tile = -1;
 275         }
 276       }
 277     }
 278
 279     // Set the last pointer to NULL
 280     job_queue_curr += -1;
 281     job_queue_curr->next = (void *)NULL;
 282
 283     // Move to the next tile
 284     job_queue += jobs_per_tile_col;
 285   }
 286
 287   for (i = 0; i < cpi->num_workers; i++) {
 288     EncWorkerData *thread_data;
 289     thread_data = &cpi->tile_thr_data[i];
 290     thread_data->thread_id = i;
 291
 292     for (tile_col = 0; tile_col < tile_cols; tile_col++)
 293       thread_data->tile_completion_status[tile_col] = 0;
 294   }
 295 }
 296
 297 int vp9_get_tiles_proc_status(MultiThreadHandle *multi_thread_ctxt,
 298                               int *tile_completion_status, int *cur_tile_id,
 299                               int tile_cols) {
 300   int tile_col;
 301   int tile_id = -1;  // Stores the tile ID with minimum proc done
 302   int max_num_jobs_remaining = 0;
 303   int num_jobs_remaining;
 304
 305   // Mark the completion to avoid check in the loop
 306   tile_completion_status[*cur_tile_id] = 1;
 307   // Check for the status of all the tiles
 308   for (tile_col = 0; tile_col < tile_cols; tile_col++) {
 309     if (tile_completion_status[tile_col] == 0) {
 310       num_jobs_remaining =
 311           vp9_get_job_queue_status(multi_thread_ctxt, tile_col);
 312       // Mark the completion to avoid checks during future switches across tiles
 313       if (num_jobs_remaining == 0) tile_completion_status[tile_col] = 1;
 314       if (num_jobs_remaining > max_num_jobs_remaining) {
 315         max_num_jobs_remaining = num_jobs_remaining;
 316         tile_id = tile_col;
 317       }
 318     }
 319   }
 320
 321   if (-1 == tile_id) {
 322     return 1;
 323   } else {
 324     // Update the cur ID to the next tile ID that will be processed,
 325     // which will be the least processed tile
 326     *cur_tile_id = tile_id;
 327     return 0;
 328   }
 329 }