| 1258 | } |
| 1259 | |
| 1260 | void TableImpl::PackSdkTasks(const std::string& server_addr, std::vector<SdkTask*>& task_list, |
| 1261 | SdkTask::TYPE task_type) { |
| 1262 | Mutex* mutex = NULL; |
| 1263 | std::map<std::string, TaskBatch*>* task_batch_map = NULL; |
| 1264 | SdkTask::TimeoutFunc task; |
| 1265 | uint64_t commit_timeout = 10000; |
| 1266 | uint32_t commit_size = commit_size_; |
| 1267 | if (task_type == SdkTask::MUTATION) { |
| 1268 | mutex = &mutation_batch_mutex_; |
| 1269 | task_batch_map = &mutation_batch_map_; |
| 1270 | commit_timeout = write_commit_timeout_; |
| 1271 | } else if (task_type == SdkTask::READ) { |
| 1272 | mutex = &reader_batch_mutex_; |
| 1273 | task_batch_map = &reader_batch_map_; |
| 1274 | commit_timeout = read_commit_timeout_; |
| 1275 | } else if (task_type == SdkTask::BATCH_MUTATION) { |
| 1276 | mutex = &mutation_batch_mutex_; |
| 1277 | task_batch_map = &mutation_batch_map_; |
| 1278 | commit_timeout = write_commit_timeout_; |
| 1279 | } else { |
| 1280 | assert(0); |
| 1281 | } |
| 1282 | |
| 1283 | TaskBatch* task_batch = NULL; |
| 1284 | bool is_instant = false; |
| 1285 | MutexLock lock(mutex); |
| 1286 | for (size_t i = 0; i < task_list.size(); ++i) { |
| 1287 | // find existing batch or create a new batch |
| 1288 | if (task_batch == NULL) { |
| 1289 | std::map<std::string, TaskBatch*>::iterator it = task_batch_map->find(server_addr); |
| 1290 | if (it != task_batch_map->end()) { |
| 1291 | task_batch = it->second; |
| 1292 | } else { |
| 1293 | task_batch = new TaskBatch; |
| 1294 | task_batch->type = task_type; |
| 1295 | task_batch->mutex = mutex; |
| 1296 | task_batch->task_batch_map = task_batch_map; |
| 1297 | task_batch->byte_size = 0; |
| 1298 | task_batch->server_addr = server_addr; |
| 1299 | task_batch->row_id_list = new std::vector<int64_t>; |
| 1300 | |
| 1301 | task_batch->SetId(next_task_id_.Inc()); |
| 1302 | (*task_batch_map)[server_addr] = task_batch; |
| 1303 | SdkTask::TimeoutFunc task = std::bind(&TableImpl::TaskBatchTimeout, this, _1); |
| 1304 | task_pool_.PutTask(task_batch, commit_timeout, task); |
| 1305 | task_batch->DecRef(); |
| 1306 | } |
| 1307 | } |
| 1308 | |
| 1309 | // put task into the batch |
| 1310 | SdkTask* sdk_task = task_list[i]; |
| 1311 | task_batch->row_id_list->push_back(sdk_task->GetId()); |
| 1312 | task_batch->byte_size += sdk_task->Size(); |
| 1313 | is_instant |= !sdk_task->IsAsync(); |
| 1314 | sdk_task->DecRef(); |
| 1315 | |
| 1316 | // commit the batch if: |
| 1317 | // 1) batch_byte_size >= max_rpc_byte_size |