| 888 | } |
| 889 | |
| 890 | void TableImpl::CommitMutations(const std::string& server_addr, |
| 891 | std::vector<RowMutationImpl*>& mu_list) { |
| 892 | tabletnode::TabletNodeClient tabletnode_client_async(thread_pool_, server_addr); |
| 893 | WriteTabletRequest* request = new WriteTabletRequest; |
| 894 | WriteTabletResponse* response = new WriteTabletResponse; |
| 895 | request->set_sequence_id(last_sequence_id_++); |
| 896 | request->set_tablet_name(name_); |
| 897 | request->set_is_sync(FLAGS_tera_sdk_write_sync); |
| 898 | |
| 899 | access_builder_->BuildRequest(request); |
| 900 | |
| 901 | bool is_instant = false; |
| 902 | std::vector<int64_t>* mu_id_list = new std::vector<int64_t>; |
| 903 | for (uint32_t i = 0; i < mu_list.size(); ++i) { |
| 904 | RowMutationImpl* row_mutation = mu_list[i]; |
| 905 | RowMutationSequence* mu_seq = request->add_row_list(); |
| 906 | if (!request->has_client_timeout_ms() || |
| 907 | (request->has_client_timeout_ms() && |
| 908 | request->client_timeout_ms() > row_mutation->TimeOut())) { |
| 909 | request->set_client_timeout_ms(row_mutation->TimeOut()); |
| 910 | } |
| 911 | mu_seq->set_row_key(row_mutation->InternalRowKey()); |
| 912 | for (uint32_t j = 0; j < row_mutation->MutationNum(); j++) { |
| 913 | const RowMutation::Mutation& mu = row_mutation->GetMutation(j); |
| 914 | tera::Mutation* mutation = mu_seq->add_mutation_sequence(); |
| 915 | SerializeMutation(mu, mutation); |
| 916 | } |
| 917 | SingleRowTxn* txn = (SingleRowTxn*)(row_mutation->GetTransaction()); |
| 918 | if (txn != NULL) { |
| 919 | txn->Serialize(mu_seq); |
| 920 | } |
| 921 | mu_id_list->push_back(row_mutation->GetId()); |
| 922 | is_instant |= !row_mutation->IsAsync(); |
| 923 | row_mutation->AddCommitTimes(); |
| 924 | row_mutation->DecRef(); |
| 925 | } |
| 926 | request->set_is_instant(is_instant); |
| 927 | |
| 928 | VLOG(20) << "commit " << mu_list.size() << " mutations to " << server_addr |
| 929 | << "timeout:" << request->client_timeout_ms(); |
| 930 | request->set_timestamp(get_micros()); |
| 931 | std::function<void(WriteTabletRequest*, WriteTabletResponse*, bool, int)> done = |
| 932 | std::bind(&TableImpl::MutateCallBackWrapper, std::weak_ptr<TableImpl>(shared_from_this()), |
| 933 | mu_id_list, _1, _2, _3, _4); |
| 934 | tabletnode_client_async.WriteTablet(request, response, done); |
| 935 | } |
| 936 | |
| 937 | void TableImpl::MutateCallBackWrapper(std::weak_ptr<TableImpl> weak_ptr_table, |
| 938 | std::vector<int64_t>* mu_id_list, WriteTabletRequest* request, |
nothing calls this directly
no test coverage detected