<commit_before>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <deque>

#include <glog/logging.h>
#include <glog/logging.h>

#include "proto/kv.pb.h"
#include "sdk/sdk.h"
#include "sdk/table_impl.h"
#include <tera.h>

DECLARE_int64(concurrent_write_handle_num);
DECLARE_int64(max_write_handle_seq);
DECLARE_int64(data_size_per_sync);

namespace mdt {

std::ostream& operator << (std::ostream& o, const FileLocation& location) {
    o << "file: " << location.fname_
      << " offset: " << location.offset_
      << " size: " << location.size_;
    return o;
}

int TableImpl::OpenTable(const std::string& db_name, const TeraOptions& tera_opt,
                         const FilesystemOptions& fs_opt, const TableDescription& table_desc,
                         Table** table_ptr) {
    const std::string& table_name = table_desc.table_name;
    // init fs adapter
    FilesystemAdapter fs_adapter;
    fs_adapter.root_path_ = fs_opt.fs_path_ + "/" + table_name + "/";
    LOG(INFO) << "data file dir: " << fs_adapter.root_path_;
    fs_adapter.env_ = fs_opt.env_;

    //init tera adapter
    TeraAdapter tera_adapter;
    tera_adapter.opt_ = tera_opt;
    tera_adapter.table_prefix_ = db_name;

    *table_ptr = new TableImpl(table_desc, tera_adapter, fs_adapter);
    assert(table_ptr);
    return 0;
}

// TableImpl ops
TableImpl::TableImpl(const TableDescription& table_desc,
                     const TeraAdapter& tera_adapter,
                     const FilesystemAdapter& fs_adapter)
    : table_desc_(table_desc),
    tera_(tera_adapter),
    fs_(fs_adapter) {
    // create fs dir
    fs_.env_->CreateDir(fs_.root_path_);

    // init write handle list
    nr_write_handle_ = (int)FLAGS_concurrent_write_handle_num;
    cur_write_handle_id_ = 0;
    write_handle_list_.clear();
    WriteHandle write_handle;
    write_handle.write_queue_.clear();
    write_handle.writer_ = NULL;
    for (int i = 0; i < nr_write_handle_; ++i) {
        write_handle_list_.push_back(write_handle);
    }

    tera::ErrorCode error_code;

    // open primary key table
    std::string primary_table_name = tera_.table_prefix_ + "#pri#" + table_desc.table_name;
    //std::string primary_table_name = GetPrimaryTable(table_desc.table_name);
    LOG(INFO) << "open primary table: " << primary_table_name;
    tera::Table* primary_table = tera_.opt_.client_->OpenTable(primary_table_name, &error_code);
    assert(primary_table);
    tera_.tera_table_map_[primary_table_name] = primary_table;

    // open index key table
    std::vector<IndexDescription>::iterator it;
    for (it = table_desc_.index_descriptor_list.begin();
         it != table_desc_.index_descriptor_list.end();
         ++it) {
        std::string index_table_name = tera_.table_prefix_ + "#" + table_desc.table_name + "#" + it->index_name;
        //std::string index_table_name = GetIndexTable(it->index_name);
        LOG(INFO) << "open index table: " << index_table_name;
        tera::Table* index_table = tera_.opt_.client_->OpenTable(index_table_name, &error_code);
        assert(index_table);
        tera_.tera_table_map_[index_table_name] = index_table;
    }
}

TableImpl::~TableImpl() {
    // free write handle
    //ReleaseDataWriter();
    // TODO: write queue release
}

/////////  batch write /////////////
struct InternalBatchPutCallbackParam {
    BatchStoreRequest* req_;
    StoreResponse* resp_;
    BatchStoreCallback user_callback_;
    void* user_param_;
    Counter counter_; // atomic counter
};

void BatchStoreRequestCallback(Table* table, StoreRequest* request,
                               StoreResponse* response,
                               void* callback_param) {
    TableImpl* table_ptr;
    table_ptr = (TableImpl*)table;
    delete request;
    InternalBatchPutCallbackParam* user_callback_param = (InternalBatchPutCallbackParam*)callback_param;
    if (user_callback_param->counter_.Dec() == 0) {
        delete response;
        user_callback_param->user_callback_(table_ptr,
                                            user_callback_param->req_,
                                            user_callback_param->resp_,
                                            user_callback_param->user_param_);
        delete user_callback_param;
    }
}

int TableImpl::Put(const BatchStoreRequest* request, StoreResponse* response,
                BatchStoreCallback callback, void* callback_param) {
    // construct vector<StoreRequest>
    std::vector<StoreRequest*> req_vec;
    std::vector<std::string>::const_iterator it = request->data_list.begin();
    for (; it != request->data_list.end(); ++it) {
        StoreRequest* req = new StoreRequest();
        req->primary_key = request->primary_key;
        req->timestamp = request->timestamp;
        req->index_list = request->index_list;
        req->data = *it;
        req_vec.push_back(req);
    }

    StoreResponse* resp = new StoreResponse();
    InternalBatchPutCallbackParam* user_callback_param = new InternalBatchPutCallbackParam();
    user_callback_param->req_ = (BatchStoreRequest*)request;
    user_callback_param->resp_ = response;
    user_callback_param->user_callback_ = callback;
    user_callback_param->user_param_ = callback_param;
    user_callback_param->counter_.Inc();

    std::vector<StoreRequest*>::iterator iter = req_vec.begin();
    for (; iter != req_vec.end(); ++iter) {
        user_callback_param->counter_.Inc();
        Put(*iter, resp, BatchStoreRequestCallback, user_callback_param);
    }

    StoreRequest* null_req = new StoreRequest();
    null_req->primary_key = "<MagicKey>";
    BatchStoreRequestCallback(this, null_req, resp, user_callback_param);
    return 0;
}

///////////  High concurrency write interface ////////////////
struct DefaultUserCallbackParam {
    CondVar* cond_;
};

void DefaultUserCallback(Table* table, StoreRequest* request,
                         StoreResponse* response,
                         void* callback_param) {
    DefaultUserCallbackParam* param = (DefaultUserCallbackParam*)callback_param;
    param->cond_->Signal();
}

// Concurrence Put interface:
//      1. wait and lock
//      2. merge request while wait
//      3. unlock, write batch
//      4. for each raw write: PutIndex()
//      5. lock, resume other write
int TableImpl::Put(const StoreRequest* req, StoreResponse* resp,
                   StoreCallback callback, void* callback_param) {
    // try to construct Default callback
    Mutex mu;
    CondVar cond(&mu);
    DefaultUserCallbackParam param;
    param.cond_ = &cond;
    if (callback == NULL) {
        callback = DefaultUserCallback;
        callback_param = &param;
    }

    // construct WriteContext
    WriteContext context(&write_mutex_);
    context.req_ = req;
    context.resp_ = resp;
    context.callback_ = callback;
    context.callback_param_ = callback_param;
    context.sync_ = true;
    context.done_ = false;

    // select write handle
    WriteHandle* write_handle = GetWriteHandle();

    // wait and lock
    write_mutex_.Lock();
    write_handle->write_queue_.push_back(&context);
    while (!context.done_ && (&context != write_handle->write_queue_.front())) {
        context.cv_.Wait();
    }
    if (context.done_) {
        write_mutex_.Unlock();
        // write finish, if sync write, just wait callback
        if (callback == DefaultUserCallback) {
            param.cond_->Wait();
        }
        return 0;
    }
        
    // merge WriteContext
    WriteBatch wb;
    std::deque<WriteContext*>::iterator iter = write_handle->write_queue_.begin();
    for (; iter != write_handle->write_queue_.end(); ++iter) {
        wb.Append(*iter);
    }

    // unlock do io
    write_mutex_.Unlock();
    VLOG(30) << ">>>>> lock put, ctx " << (uint64_t)(&context);

    // batch write file system
    FileLocation location;
    DataWriter* writer = GetDataWriter(write_handle);
    writer->AddRecord(wb.rep_, &location);

    // write index table
    std::vector<WriteContext*>::iterator it = wb.context_list_.begin();
    for (; it != wb.context_list_.end(); ++it) {<commit_msg> #33 : high concurrency put<commit_after>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <deque>

#include <glog/logging.h>
#include <glog/logging.h>

#include "proto/kv.pb.h"
#include "sdk/sdk.h"
#include "sdk/table_impl.h"
#include <tera.h>

DECLARE_int64(concurrent_write_handle_num);
DECLARE_int64(max_write_handle_seq);
DECLARE_int64(data_size_per_sync);

namespace mdt {

std::ostream& operator << (std::ostream& o, const FileLocation& location) {
    o << "file: " << location.fname_
      << " offset: " << location.offset_
      << " size: " << location.size_;
    return o;
}

int TableImpl::OpenTable(const std::string& db_name, const TeraOptions& tera_opt,
                         const FilesystemOptions& fs_opt, const TableDescription& table_desc,
                         Table** table_ptr) {
    const std::string& table_name = table_desc.table_name;
    // init fs adapter
    FilesystemAdapter fs_adapter;
    fs_adapter.root_path_ = fs_opt.fs_path_ + "/" + table_name + "/";
    LOG(INFO) << "data file dir: " << fs_adapter.root_path_;
    fs_adapter.env_ = fs_opt.env_;

    //init tera adapter
    TeraAdapter tera_adapter;
    tera_adapter.opt_ = tera_opt;
    tera_adapter.table_prefix_ = db_name;

    *table_ptr = new TableImpl(table_desc, tera_adapter, fs_adapter);
    assert(table_ptr);
    return 0;
}

// TableImpl ops
TableImpl::TableImpl(const TableDescription& table_desc,
                     const TeraAdapter& tera_adapter,
                     const FilesystemAdapter& fs_adapter)
    : table_desc_(table_desc),
    tera_(tera_adapter),
    fs_(fs_adapter) {
    // create fs dir
    fs_.env_->CreateDir(fs_.root_path_);

    // init write handle list
    nr_write_handle_ = (int)FLAGS_concurrent_write_handle_num;
    cur_write_handle_id_ = 0;
    write_handle_list_.clear();
    WriteHandle write_handle;
    write_handle.write_queue_.clear();
    write_handle.writer_ = NULL;
    for (int i = 0; i < nr_write_handle_; ++i) {
        write_handle_list_.push_back(write_handle);
    }

    tera::ErrorCode error_code;

    // open primary key table
    std::string primary_table_name = tera_.table_prefix_ + "#pri#" + table_desc.table_name;
    //std::string primary_table_name = GetPrimaryTable(table_desc.table_name);
    LOG(INFO) << "open primary table: " << primary_table_name;
    tera::Table* primary_table = tera_.opt_.client_->OpenTable(primary_table_name, &error_code);
    assert(primary_table);
    tera_.tera_table_map_[primary_table_name] = primary_table;

    // open index key table
    std::vector<IndexDescription>::iterator it;
    for (it = table_desc_.index_descriptor_list.begin();
         it != table_desc_.index_descriptor_list.end();
         ++it) {
        std::string index_table_name = tera_.table_prefix_ + "#" + table_desc.table_name + "#" + it->index_name;
        //std::string index_table_name = GetIndexTable(it->index_name);
        LOG(INFO) << "open index table: " << index_table_name;
        tera::Table* index_table = tera_.opt_.client_->OpenTable(index_table_name, &error_code);
        assert(index_table);
        tera_.tera_table_map_[index_table_name] = index_table;
    }
}

TableImpl::~TableImpl() {
    // free write handle
    //ReleaseDataWriter();
    // TODO: write queue release
}

/////////  batch write /////////////
struct InternalBatchPutCallbackParam {
    BatchStoreRequest* req_;
    StoreResponse* resp_;
    BatchStoreCallback user_callback_;
    void* user_param_;
    Counter counter_; // atomic counter
};

void BatchStoreRequestCallback(Table* table, StoreRequest* request,
                               StoreResponse* response,
                               void* callback_param) {
    TableImpl* table_ptr;
    table_ptr = (TableImpl*)table;
    delete request;
    InternalBatchPutCallbackParam* user_callback_param = (InternalBatchPutCallbackParam*)callback_param;
    if (user_callback_param->counter_.Dec() == 0) {
        delete response;
        user_callback_param->user_callback_(table_ptr,
                                            user_callback_param->req_,
                                            user_callback_param->resp_,
                                            user_callback_param->user_param_);
        delete user_callback_param;
    }
}

int TableImpl::Put(const BatchStoreRequest* request, StoreResponse* response,
                BatchStoreCallback callback, void* callback_param) {
    // construct vector<StoreRequest>
    std::vector<StoreRequest*> req_vec;
    std::vector<std::string>::const_iterator it = request->data_list.begin();
    for (; it != request->data_list.end(); ++it) {
        StoreRequest* req = new StoreRequest();
        req->primary_key = request->primary_key;
        req->timestamp = request->timestamp;
        req->index_list = request->index_list;
        req->data = *it;
        req_vec.push_back(req);
    }

    StoreResponse* resp = new StoreResponse();
    InternalBatchPutCallbackParam* user_callback_param = new InternalBatchPutCallbackParam();
    user_callback_param->req_ = (BatchStoreRequest*)request;
    user_callback_param->resp_ = response;
    user_callback_param->user_callback_ = callback;
    user_callback_param->user_param_ = callback_param;
    user_callback_param->counter_.Inc();

    std::vector<StoreRequest*>::iterator iter = req_vec.begin();
    for (; iter != req_vec.end(); ++iter) {
        user_callback_param->counter_.Inc();
        Put(*iter, resp, BatchStoreRequestCallback, user_callback_param);
    }

    StoreRequest* null_req = new StoreRequest();
    null_req->primary_key = "<MagicKey>";
    BatchStoreRequestCallback(this, null_req, resp, user_callback_param);
    return 0;
}

///////////  High concurrency write interface ////////////////
struct DefaultUserCallbackParam {
    CondVar* cond_;
};

void DefaultUserCallback(Table* table, StoreRequest* request,
                         StoreResponse* response,
                         void* callback_param) {
    DefaultUserCallbackParam* param = (DefaultUserCallbackParam*)callback_param;
    param->cond_->Signal();
}

// Concurrence Put interface:
//      1. wait and lock
//      2. merge request while wait
//      3. unlock, write batch
//      4. for each raw write: PutIndex()
//      5. lock, resume other write
int TableImpl::Put(const StoreRequest* req, StoreResponse* resp,
                   StoreCallback callback, void* callback_param) {
    // try to construct Default callback
    Mutex mu;
    CondVar cond(&mu);
    DefaultUserCallbackParam param;
    param.cond_ = &cond;
    if (callback == NULL) {
        callback = DefaultUserCallback;
        callback_param = &param;
    }

    // construct WriteContext
    WriteContext context(&write_mutex_);
    context.req_ = req;
    context.resp_ = resp;
    context.callback_ = callback;
    context.callback_param_ = callback_param;
    context.sync_ = true;
    context.done_ = false;

    // select write handle
    WriteHandle* write_handle = GetWriteHandle();

    // wait and lock
    write_mutex_.Lock();
    write_handle->write_queue_.push_back(&context);
    while (!context.done_ && (&context != write_handle->write_queue_.front())) {
        context.cv_.Wait();
    }
    if (context.done_) {
        write_mutex_.Unlock();
        // write finish, if sync write, just wait callback
        if (callback == DefaultUserCallback) {
            param.cond_->Wait();
        }
        return 0;
    }

    // merge WriteContext
    WriteBatch wb;
    std::deque<WriteContext*>::iterator iter = write_handle->write_queue_.begin();
    for (; iter != write_handle->write_queue_.end(); ++iter) {
        wb.Append(*iter);
    }

    // unlock do io
    write_mutex_.Unlock();
    VLOG(30) << ">>>>> lock put, ctx " << (uint64_t)(&context);

    // batch write file system
    FileLocation location;
    DataWriter* writer = GetDataWriter(write_handle);
    writer->AddRecord(wb.rep_, &location);

    // write index table
    std::vector<WriteContext*>::iterator it = wb.context_list_.begin();
    for (; it != wb.context_list_.end(); ++it) {
{'fraction_non_alphanumeric': 0.11946848713885164, 'fraction_numerical': 0.003169572107765452, 'mean_word_length': 2.945887445887446, 'pattern_counts': {'":': 0, '<': 89, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '621574', 'n_tokens_mistral': 4891, 'n_tokens_neox': 4766, 'n_words': 1024}