Method ComputeAsync

tf_euler/kernels/get_binary_feature_op.cc:60–115

};

void GetBinaryFeature::ComputeAsync(OpKernelContext* ctx, DoneCallback done) {
  auto nodes = ctx->input(0);
  auto& shape = nodes.shape();

  std::vector<Tensor*> outputs(N_, nullptr);
  TensorShape output_shape;
  output_shape.AddDim(shape.dim_size(0));
  for (auto i = 0; i < N_; ++i) {
    OP_REQUIRES_OK(ctx, ctx->allocate_output(i, output_shape, &outputs[i]));
  }

  auto nodes_flat = nodes.flat<int64>();
  size_t nodes_size = nodes_flat.size();

  auto query = new euler::Query(query_str_);
  auto t_nodes = query->AllocInput("nodes", {nodes_size}, euler::kUInt64);
  for (size_t i = 0; i < feature_names_.size(); ++i) {
    auto t_fid = query->AllocInput("__" + feature_names_[i], {1},
                                   euler::kString);
    *(t_fid->Raw<std::string*>()[0]) = "binary_" + feature_names_[i];
  }
  std::copy(nodes_flat.data(), nodes_flat.data() + nodes_flat.size(),
            t_nodes->Raw<int64_t>());

  auto callback = [outputs, done, query, nodes_size, this]() {
    auto results_map = query->GetResult(res_names_);
    std::stringstream ss;
    for (size_t i = 0 ; i < feature_names_.size(); ++i) {
      ss.str("");
      ss << "fea:" << i * 2;
      std::string fea_idx = ss.str();

      ss.str("");
      ss << "fea:" << i * 2 + 1;
      std::string fea_val = ss.str();

      if (results_map[fea_idx]->NumElements() != nodes_size * 2) {
        EULER_LOG(FATAL) << "Binary Feature Result Index Num Error:" <<
            results_map[fea_idx]->NumElements() << "Expect: " << nodes_size * 2;
      }
      for (size_t j = 0; j < nodes_size; ++j) {
        size_t start = results_map[fea_idx]->Raw<int32_t>()[j * 2];
        size_t end = results_map[fea_idx]->Raw<int32_t>()[j * 2 + 1];
        auto data = outputs[i]->flat<tensorflow::string>();
        std::string f_v(end - start, 0);
        std::copy(results_map[fea_val]->Raw<char>() + start,
                  results_map[fea_val]->Raw<char>() + end,
                  f_v.begin());
        data(j) = f_v;
      }
    }
    delete query;
    done();
  };
  euler::QueryProxy::GetInstance()->RunAsyncGremlin(query, callback);
}

REGISTER_KERNEL_BUILDER(