| 219 | |
| 220 | template <typename OID_T, typename VID_T, typename VD_T, typename ED_T> |
| 221 | void RunGraphX(vineyard::Client& client, grape::CommSpec& comm_spec, |
| 222 | std::string& frag_name) { |
| 223 | using ProjectedFragmentType = |
| 224 | typename gs::ArrowProjectedFragment<OID_T, VID_T, VD_T, ED_T>; |
| 225 | boost::property_tree::ptree pt = flags2Ptree(); |
| 226 | |
| 227 | auto pair = splitAndGet(comm_spec, FLAGS_frag_ids); |
| 228 | auto cur_frag_id = pair.first; |
| 229 | auto graphx_pid_id = pair.second; |
| 230 | LOG(INFO) << " graphx pid: " << graphx_pid_id << " fid " << comm_spec.fid() |
| 231 | << " frag id " << cur_frag_id; |
| 232 | |
| 233 | std::shared_ptr<ProjectedFragmentType> fragment = |
| 234 | std::dynamic_pointer_cast<ProjectedFragmentType>( |
| 235 | client.GetObject(cur_frag_id)); |
| 236 | |
| 237 | { |
| 238 | std::vector<int32_t> worker_id_to_fid; |
| 239 | worker_id_to_fid.resize(comm_spec.fnum()); |
| 240 | auto fid = fragment->fid(); |
| 241 | MPI_Allgather(&fid, 1, MPI_INT, worker_id_to_fid.data(), 1, MPI_INT, |
| 242 | comm_spec.comm()); |
| 243 | std::stringstream ss; |
| 244 | for (grape::fid_t i = 0; i < comm_spec.fnum(); ++i) { |
| 245 | ss << ";" << i << ":" << worker_id_to_fid[i]; |
| 246 | } |
| 247 | auto worker_id_to_fid_str = ss.str().substr(1); |
| 248 | // LOG(INFO) << "worker_id_to_fid_str: " << worker_id_to_fid_str; |
| 249 | pt.put("worker_id_to_fid", worker_id_to_fid_str); |
| 250 | } |
| 251 | pt.put("frag_name", frag_name); |
| 252 | if (getenv("USER_JAR_PATH")) { |
| 253 | pt.put("jar_name", getenv("USER_JAR_PATH")); |
| 254 | } else { |
| 255 | LOG(ERROR) << "USER_JAR_PATH not set"; |
| 256 | return; |
| 257 | } |
| 258 | // get params str |
| 259 | std::stringstream ss; |
| 260 | boost::property_tree::json_parser::write_json(ss, pt); |
| 261 | std::string new_params = ss.str(); |
| 262 | |
| 263 | double t0 = grape::GetCurrentTime(); |
| 264 | |
| 265 | for (int i = 0; i < 1; ++i) { |
| 266 | if (FLAGS_context_class == |
| 267 | "com.alibaba.graphscope.context.GraphXParallelAdaptorContext") { |
| 268 | using APP_TYPE = JavaPIEProjectedParallelAppIE<ProjectedFragmentType>; |
| 269 | LOG(INFO) << "Message strategy: " |
| 270 | << (APP_TYPE::message_strategy == |
| 271 | grape::MessageStrategy::kAlongIncomingEdgeToOuterVertex); |
| 272 | Query<ProjectedFragmentType, APP_TYPE>(comm_spec, fragment, new_params, |
| 273 | FLAGS_user_lib_path); |
| 274 | } else { |
| 275 | LOG(ERROR) << "Not recognized context clz" << FLAGS_context_class; |
| 276 | } |
| 277 | } |
| 278 | double t1 = grape::GetCurrentTime(); |