Add ImportKV service #2881

huachaohuang · 2018-03-28T15:42:29Z

Implement Open/Write/Close APIs to write key-value pairs to import
server.

Open opens an engine identified by an UUID.

Write creates a write stream to write key-value batch to the opened
engine. Different clients can write to the same engine concurrently.

Close closes the engine after all write batches are finished. An
engine can only be closed when all write streams are closed. An engine
can only be closed once, and it can not be opened again once it is
closed.

After these three steps, the data in the engine is ready to be
imported to TiKV, but the Import API is not included in this PR.

breezewish · 2018-03-30T05:48:34Z

in concept, what's the relationship of engine, engineFile, client, token, uuid?

huachaohuang · 2018-03-30T06:50:19Z

Engine: A storage engine to store key-value pairs.
EngineDir/EngineFile: Just simple wrappers to manage different engines in a directory.
UUID: The ID of an engine, different clients can operate on the same engine concurrently.
Token: The ID of a client, and it will be bound to an UUID to write to the engine in streaming.

breezewish

rest LGTM

breezewish · 2018-03-30T07:09:02Z

src/import/kv_importer.rs

+
+pub struct KVImporter {
+    dir: EngineDir,
+    token: AtomicUsize,


I suggest to rename to a more obvious name, i.e. clientId

We use this token name in different places in TiKV, so I think it is fine.

breezewish · 2018-03-30T07:10:31Z

src/import/kv_importer.rs

+        self.token.fetch_add(1, Ordering::SeqCst)
+    }
+
+    fn insert(&self, token: Token, engine: Arc<EngineFile>) {


The function name is misleading. It can be easily mis-interpreted as something related to the KvEngine's data insert.

Well, HashMap has an insert method too. KVImporter is a struct to maintain the relationship between clients and engines, it has nothing to do with data insert/remove.

For a HashMap's insert, we know exactly what it inserts. When it is interpreted as "insert into HashMap" the purpose is obvious. However, for KVImporter's insert, usually we have no idea unless take a look at the implementation. I mean, "insert into KVImporter" is confusing, because we neither know what KVImporter really does, nor what insert really does, according to their names. I suggest to either document as comment for at least one of them, or change the name of the function so that it is self-explaining.

breezewish · 2018-03-30T07:10:42Z

src/import/kv_importer.rs

+        assert!(inner.clients.insert(token, engine).is_none());
+    }
+
+    pub fn remove(&self, token: Token) -> Option<Arc<EngineFile>> {


ditto. The function name is misleading.

breezewish · 2018-03-30T07:15:42Z

src/import/kv_importer.rs

+    }
+
+    pub fn write(&self, token: Token, batch: WriteBatch) -> Result<usize> {
+        match self.remove(token) {


It's wired that write contains a detach operation.

We remove it here so we don't need to lock when writing to the engine.

Is it an implication that a token can be only used to write once?

No, a token can be used to write multiple times, that's why we insert it back after a successful write.

breezewish · 2018-03-30T07:18:43Z

src/import/kv_importer.rs

+                Ok(engine) => engine,
+                Err(engine) => {
+                    inner.engines.insert(uuid, engine);
+                    return Err(Error::EngineInUse(uuid));


The usage of engine is actually maintained by whether or not there are references to the engine. However semantically they are not exactly the same concept. Are there better solutions?

Well, it is exactly the same concept. When a client writes to it, it acquires a reference, and when the client finishes the write, it releases the reference. So if there are no references to the engine, there is no one using the engine, that's exactly what I want to check here.

How can you ensure that "when the client finishes the write, it releases the reference"? If this is a guarantee according to implementations, I think it is somewhat weak because of no strong restrictions. If it is a guarantee by interface I accept.

When a client finishes a write stream or any errors occur, the stream will be closed and the reference will be released. And if the client doesn't close all streams, it can't close the engine, that's a protocol between the client and the server documented here https://github.com/huachaohuang/kvproto/blob/import-1/proto/importpb.proto#L22.

if we can ensure the client uses the unique engine with UUID, so here only the same client can close the engine. If the client closes the engine, the following write will be failed, I think it is acceptable. So I think here we can close the engine directly, no need to check in used any more.

Not really. Different clients from different machines can write to the same engine concurrently, and any client can close the engine if it knows that all writes are done. Moreover, I must use the unsafe way to close the engine with an Arc if I don't check the reference.

huachaohuang · 2018-04-02T05:33:29Z

@breeswish I have changed the function name and add tests, PTAL thanks.

huachaohuang · 2018-04-02T07:18:29Z

@zhangjinpeng1987 PTAL

breezewish

rest LGTM

breezewish · 2018-04-02T09:29:52Z

Cargo.toml

@@ -82,7 +82,8 @@ git = "https://github.com/pingcap/murmur3.git"
 git = "https://github.com/pingcap/rust-rocksdb.git"

 [dependencies.kvproto]
-git = "https://github.com/pingcap/kvproto.git"
+git = "https://github.com/huachaohuang/kvproto.git"


Needs update?

Wait until I get two LGTMs.

breezewish · 2018-04-02T09:31:15Z

etc/config-template.toml

@@ -442,6 +442,7 @@
 # key-path = ""

 [import]
+# import-dir = "/tmp/tikv/import"


Please add a comment for this field.

breezewish · 2018-04-02T09:34:55Z

src/import/kv_importer.rs

+    pub fn open_engine(&self, uuid: Uuid) -> Result<()> {
+        let mut inner = self.inner.lock().unwrap();
+        if inner.engines.contains_key(&uuid) {
+            return Ok(());


Is it proper to return Ok if the engine is already opened using a uuid?

Yes, clients may want to open the engine concurrently.

siddontang · 2018-04-03T02:18:12Z

src/import/engine.rs

+impl Engine {
+    pub fn new<P: AsRef<Path>>(path: P, uuid: Uuid, opts: DbConfig) -> Result<Engine> {
+        let db = {
+            let (db_opts, cf_opts) = tune_dboptions_for_bulk_load(&opts);


do you only import to the default CF?

No, the data for the write CF will be generated from the default CF when we actually import the SST files.

so here no need to change the Write CF options?

Yes, no need.

siddontang · 2018-04-03T06:07:29Z

src/import/engine.rs

+    }
+
+    pub fn write(&self, mut batch: WriteBatch) -> Result<usize> {
+        let wb = RawBatch::new();


can you new with a predefined capacity?

siddontang · 2018-04-03T06:09:40Z

src/import/kv_importer.rs

+
+    /// Write to the attached engine.
+    pub fn write_engine(&self, token: Token, batch: WriteBatch) -> Result<usize> {
+        match self.remove_client(token) {


why remove and insert client every time?

Avoid locking while the client is writing to the engine.

overvenus · 2018-04-03T06:51:40Z

src/import/kv_server.rs

+
+use super::{ImportKVService, KVImporter};
+
+const MAX_GRPC_MSG_LEN: usize = 32 * 1024 * 1024;


Should it be configurable?

Doesn't seem like necessary for now.

overvenus · 2018-04-03T06:53:29Z

src/import/kv_service.rs

+                            let batch_size = import1.write_engine(token, batch)?;
+                            IMPORT_WRITE_CHUNK_BYTES.observe(batch_size as f64);
+                        }
+                        IMPORT_WRITE_CHUNK_DURATION.observe(start.elapsed_secs());


Histogram has timers.

Yep, but I only want to record the time of successful writes.

overvenus · 2018-04-03T07:18:11Z

src/import/kv_importer.rs

+    }
+
+    /// Write to the attached engine.
+    pub fn write_engine(&self, token: Token, batch: WriteBatch) -> Result<usize> {


How about get_engine? So we can avoid lock-removal and lock-insertion over and over again.

Any hint about how to do that? I have no idea how to carry the engine outside.

Basically, I think it should look like:

let engine = { let mut inner = self.inner.lock().unwrap(); inner.clients.get(&token).clone() } stream.fold(engine, |engine| { ... });

Thanks for your advice, the code looks simpler now, PTAL.

overvenus · 2018-04-03T07:22:20Z

src/bin/tikv-server.rs

@@ -299,6 +299,15 @@ fn run_raft_server(pd_client: RpcClient, cfg: &TiKvConfig, security_mgr: Arc<Sec
    }
 }

+fn run_import_server(cfg: &TiKvConfig) {


You may want enable mertics.

overvenus · 2018-04-03T07:25:02Z

src/import/config.rs

    pub num_threads: usize,
    pub stream_channel_window: usize,
 }

 impl Default for Config {
    fn default() -> Config {
        Config {
+            import_dir: "/tmp/tikv/import".to_owned(),


Why not data-dir/import?

There is no data-dir here, and it is conceptually not related to the tikv data-dir.

overvenus · 2018-04-03T07:33:59Z

src/import/engine.rs

+        let wb = RawBatch::with_capacity(wb_cap);
+        let commit_ts = batch.get_commit_ts();
+        for m in batch.take_mutations().iter_mut() {
+            let k = Key::from_raw(m.get_key()).append_ts(commit_ts);


Could you reuse the k?

What's that mean? The key is different every time.

Reuse its alloction.

It doesn't seem we have a way to reuse it now, we can optimize this in the future when necessary.

siddontang · 2018-04-03T14:08:57Z

src/bin/tikv-server.rs

@@ -350,7 +363,7 @@ fn overwrite_config_with_cmd_args(config: &mut TiKvConfig, matches: &ArgMatches)
        config.raft_store.capacity = capacity;
    }

-    if matches.is_present("import-mode") {
+    if matches.is_present("import-mode") || matches.is_present("import-server") {


what is the different between import-mode and import-server?

import-mode tunes some configs automatically, while import-server runs as an import server, which has nothing to do with tikv. Don't pay too much attention here now, I will try to move this to other place and build import server as a standalone binary in next pr.

overvenus

LGTM

huachaohuang · 2018-04-04T05:10:27Z

/run-integration-tests

Implement Open/Write/Close APIs to write key-value pairs to import server. `Open` opens an engine identified by an UUID. `Write` creates a write stream to write key-value batch to the opened engine. Different clients can write to the same engine concurrently. `Close` closes the engine after all write batches are finished. An engine can only be closed when all write streams are closed. An engine can only be closed once, and it can not be opened again once it is closed. After these three steps, the data in the engine is ready to be imported to TiKV, but the `Import` API is not included in this PR.

huachaohuang · 2018-04-04T07:26:58Z

/run-integration-tests

Implement Open/Write/Close APIs to write key-value pairs to import server. `Open` opens an engine identified by an UUID. `Write` creates a write stream to write key-value batch to the opened engine. Different clients can write to the same engine concurrently. `Close` closes the engine after all write batches are finished. An engine can only be closed when all write streams are closed. An engine can only be closed once, and it can not be opened again once it is closed. After these three steps, the data in the engine is ready to be imported to TiKV, but the `Import` API is not included in this PR.

huachaohuang requested review from breezewish and zhangjinpeng87 March 28, 2018 15:42

breezewish reviewed Mar 30, 2018

View reviewed changes

siddontang added the priority/critical Priority: Critical label Apr 2, 2018

breezewish reviewed Apr 2, 2018

View reviewed changes

siddontang reviewed Apr 3, 2018

View reviewed changes

tikv deleted a comment from breezewish Apr 3, 2018

overvenus reviewed Apr 3, 2018

View reviewed changes

siddontang reviewed Apr 3, 2018

View reviewed changes

overvenus reviewed Apr 4, 2018

View reviewed changes

huachaohuang added 2 commits April 4, 2018 15:26

Fix rebase

1c2ab25

overvenus approved these changes Apr 4, 2018

View reviewed changes

huachaohuang merged commit 6f5e9d2 into tikv:master Apr 4, 2018

huachaohuang deleted the import-1 branch April 4, 2018 09:30


		use super::{ImportKVService, KVImporter};

		const MAX_GRPC_MSG_LEN: usize = 32 * 1024 * 1024;

Add ImportKV service #2881

Add ImportKV service #2881

Conversation

huachaohuang commented Mar 28, 2018

breezewish commented Mar 30, 2018 • edited Loading

huachaohuang commented Mar 30, 2018

breezewish left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

breezewish Apr 2, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huachaohuang commented Apr 2, 2018

huachaohuang commented Apr 2, 2018

breezewish left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huachaohuang Apr 3, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

siddontang Apr 3, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

overvenus left a comment

Choose a reason for hiding this comment

huachaohuang commented Apr 4, 2018

huachaohuang commented Apr 4, 2018

breezewish commented Mar 30, 2018 •

edited

Loading

breezewish Apr 2, 2018 •

edited

Loading

huachaohuang Apr 3, 2018 •

edited

Loading

siddontang Apr 3, 2018 •

edited

Loading