| 2556 | } |
| 2557 | |
| 2558 | void TabletImpl::SendData(RpcController* controller, |
| 2559 | const ::fedb::api::SendDataRequest* request, |
| 2560 | ::fedb::api::GeneralResponse* response, |
| 2561 | Closure* done) { |
| 2562 | brpc::ClosureGuard done_guard(done); |
| 2563 | brpc::Controller* cntl = static_cast<brpc::Controller*>(controller); |
| 2564 | uint32_t tid = request->tid(); |
| 2565 | uint32_t pid = request->pid(); |
| 2566 | std::string db_root_path; |
| 2567 | bool ok = ChooseDBRootPath(tid, pid, db_root_path); |
| 2568 | if (!ok) { |
| 2569 | response->set_code(::fedb::base::ReturnCode::kFailToGetDbRootPath); |
| 2570 | response->set_msg("fail to get db root path"); |
| 2571 | PDLOG(WARNING, "fail to get table db root path for tid %u, pid %u", tid, |
| 2572 | pid); |
| 2573 | return; |
| 2574 | } |
| 2575 | std::string combine_key = std::to_string(tid) + "_" + std::to_string(pid) + |
| 2576 | "_" + request->file_name(); |
| 2577 | std::shared_ptr<FileReceiver> receiver; |
| 2578 | std::shared_ptr<Table> table; |
| 2579 | if (request->block_id() == 0) { |
| 2580 | table = GetTable(tid, pid); |
| 2581 | } |
| 2582 | { |
| 2583 | std::lock_guard<std::mutex> lock(mu_); |
| 2584 | auto iter = file_receiver_map_.find(combine_key); |
| 2585 | if (request->block_id() == 0) { |
| 2586 | if (table && request->dir_name() != "index") { |
| 2587 | PDLOG(WARNING, "table already exists. tid %u, pid %u", tid, |
| 2588 | pid); |
| 2589 | response->set_code( |
| 2590 | ::fedb::base::ReturnCode::kTableAlreadyExists); |
| 2591 | response->set_msg("table already exists"); |
| 2592 | return; |
| 2593 | } |
| 2594 | if (iter == file_receiver_map_.end()) { |
| 2595 | std::string path = db_root_path + "/" + std::to_string(tid) + |
| 2596 | "_" + std::to_string(pid) + "/"; |
| 2597 | std::string dir_name; |
| 2598 | if (request->has_dir_name() && request->dir_name().size() > 0) { |
| 2599 | dir_name = request->dir_name(); |
| 2600 | if (dir_name != "index") { |
| 2601 | path.append("snapshot/"); |
| 2602 | } |
| 2603 | path.append(request->dir_name() + "/"); |
| 2604 | } else if (request->file_name() != "table_meta.txt") { |
| 2605 | path.append("snapshot/"); |
| 2606 | } |
| 2607 | file_receiver_map_.insert(std::make_pair( |
| 2608 | combine_key, std::make_shared<FileReceiver>( |
| 2609 | request->file_name(), dir_name, path))); |
| 2610 | iter = file_receiver_map_.find(combine_key); |
| 2611 | } |
| 2612 | if (!iter->second->Init()) { |
| 2613 | PDLOG(WARNING, |
| 2614 | "file receiver init failed. tid %u, pid %u, file_name %s", |
| 2615 | tid, pid, request->file_name().c_str()); |