zan-extension/src/workers/zanTaskWorker.c
// Send the task result to worker
Send the task result to worker
Send the task result to worker
int zanTaskworker_finish(char *data, int data_len, int flags) {
  zanDebug("data=%s, flags=%d, worker_id=%d, type=%d", data, flags,
           ServerWG.worker_id, ServerG.process_type);
  zanServer *serv = ServerG.serv;
  if (ServerG.servSet.task_worker_num < 1) {
    zanWarn("cannot use task/finish, because no set task_worker_num.");
    return ZAN_ERR;
  }
  uint16_t source_worker_id = current_task->info.from_id;
  zanWorker *worker = zanServer_get_worker(serv, source_worker_id);
  int ret = 0;
  swEventData buf;
  if (swTask_type(current_task) & ZAN_TASK_NONBLOCK) {
    buf.info.type = SW_EVENT_FINISH;
    buf.info.fd = current_task->info.fd;
    swTask_type(&buf) = flags;
    if (data_len >= SW_IPC_MAX_SIZE - sizeof(buf.info)) {
      if (zanTaskWorker_largepack(&buf, data, data_len) < 0) {
        zanWarn("large task pack failed()");
        return SW_ERR;
      }
    } else {
      memcpy(buf.data, data, data_len);
      buf.info.len = data_len;
    }
    ret = zanWorker_send2worker(worker, &buf, sizeof(buf.info) + buf.info.len,
                                ZAN_PIPE_MASTER);
  } else {
    zanError("error task type=%d, flags=%d, cur_worker_id=%d",
             swTask_type(current_task), flags, ServerWG.worker_id);
  }
  if (ret < 0) {
    zanError("TaskWorker: send result to worker failed.");
  }
  return ret;
}
int zanTaskworker_finish ( char * data , int data_len , int flags ) { zanDebug ( " " , data , flags , ServerWG . worker_id , ServerG . process_type ) ; zanServer * serv = ServerG . serv ; if ( ServerG . servSet . task_worker_num < 1 ) { zanWarn ( " " ) ; return ZAN_ERR ; } uint16_t source_worker_id = current_task -> info . from_id ; zanWorker * worker = zanServer_get_worker ( serv , source_worker_id ) ; int ret = 0 ; swEventData buf ; if ( swTask_type ( current_task ) & ZAN_TASK_NONBLOCK ) { buf . info . type = SW_EVENT_FINISH ; buf . info . fd = current_task -> info . fd ; swTask_type ( & buf ) = flags ; if ( data_len >= SW_IPC_MAX_SIZE - sizeof ( buf . info ) ) { if ( zanTaskWorker_largepack ( & buf , data , data_len ) < 0 ) { zanWarn ( " " ) ; return SW_ERR ; } } else { memcpy ( buf . data , data , data_len ) ; buf . info . len = data_len ; } ret = zanWorker_send2worker ( worker , & buf , sizeof ( buf . info ) + buf . info . len , ZAN_PIPE_MASTER ) ; } else { zanError ( " " , swTask_type ( current_task ) , flags , ServerWG . worker_id ) ; } if ( ret < 0 ) { zanError ( " " ) ; } return ret ; }
Send the task result to worker
Send the task result to worker
// for swoole_server_task // write to file