From 15a36d76e81a9e066f20435e23a149bfbdd6d936 Mon Sep 17 00:00:00 2001 From: Stephen Carman Date: Fri, 23 Aug 2019 13:27:28 -0400 Subject: [PATCH] Most changes done have to work out some Send + Sync issues and flush route --- Cargo.lock | 91 ++++++++++++++++++++++---------------------- src/bin/toshi.rs | 25 ++++++------ src/commit.rs | 34 ++++++++++++----- src/handlers/bulk.rs | 31 +++++++++------ src/router.rs | 14 +++++-- 5 files changed, 112 insertions(+), 83 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4c072703..63b39961 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -63,12 +63,12 @@ dependencies = [ [[package]] name = "autocfg" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "backtrace" -version = "0.3.34" +version = "0.3.35" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "backtrace-sys 0.1.31 (registry+https://github.com/rust-lang/crates.io-index)", @@ -180,7 +180,7 @@ dependencies = [ "bitflags 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)", "textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)", - "unicode-width 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "unicode-width 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "vec_map 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -197,7 +197,7 @@ name = "codegen" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "indexmap 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -357,8 +357,8 @@ name = "ctor" version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "quote 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -396,7 +396,7 @@ name = "failure" version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "backtrace 0.3.34 (registry+https://github.com/rust-lang/crates.io-index)", + "backtrace 0.3.35 (registry+https://github.com/rust-lang/crates.io-index)", "failure_derive 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -500,7 +500,7 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.1.9" +version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "cfg-if 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)", @@ -518,7 +518,7 @@ dependencies = [ "fnv 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", "futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)", "http 0.1.18 (registry+https://github.com/rust-lang/crates.io-index)", - "indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "indexmap 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", "string 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -531,7 +531,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "ahash 0.2.12 (registry+https://github.com/rust-lang/crates.io-index)", - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -632,7 +632,7 @@ dependencies = [ [[package]] name = "indexmap" -version = "1.0.2" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] @@ -923,7 +923,7 @@ name = "num-integer" version = "0.1.41" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "num-traits 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -940,7 +940,7 @@ name = "num-traits" version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1090,8 +1090,8 @@ version = "0.5.9" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "quote 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1168,7 +1168,7 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.1" +version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1212,7 +1212,7 @@ name = "rand" version = "0.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)", "rand_chacha 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "rand_core 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1230,7 +1230,7 @@ name = "rand" version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "getrandom 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)", + "getrandom 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)", "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "rand_chacha 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1243,7 +1243,7 @@ name = "rand_chacha" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "rand_core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -1274,7 +1274,7 @@ name = "rand_core" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "getrandom 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)", + "getrandom 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1329,7 +1329,7 @@ name = "rand_pcg" version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "rand_core 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -1521,8 +1521,8 @@ version = "1.0.99" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "quote 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1626,11 +1626,11 @@ dependencies = [ [[package]] name = "syn" -version = "1.0.2" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", - "quote 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", "unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -1730,7 +1730,7 @@ name = "textwrap" version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "unicode-width 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "unicode-width 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1989,7 +1989,7 @@ dependencies = [ "tower-grpc 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "tower-hyper 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "tower-request-modifier 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", - "tracing 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "tracing 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "tracing-fmt 0.0.1-alpha.3 (registry+https://github.com/rust-lang/crates.io-index)", "tracing-futures 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)", "tracing-log 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2017,7 +2017,7 @@ dependencies = [ "http 0.1.18 (registry+https://github.com/rust-lang/crates.io-index)", "hyper 0.12.33 (registry+https://github.com/rust-lang/crates.io-index)", "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)", - "tracing 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "tracing 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -2047,7 +2047,7 @@ dependencies = [ "tokio-sync 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", - "tracing 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "tracing 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -2220,24 +2220,23 @@ dependencies = [ [[package]] name = "tracing" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "cfg-if 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "spin 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)", - "tracing-attributes 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "tracing-attributes 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", "tracing-core 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] name = "tracing-attributes" -version = "0.1.1" +version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)", - "quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -2272,7 +2271,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)", "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)", - "tracing 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", + "tracing 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -2327,7 +2326,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "unicode-width" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] @@ -2497,8 +2496,8 @@ dependencies = [ "checksum ascii 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)" = "91e320562a8fa3286a481b7189f89578ace6b20df99e123c87f2f509c957c5d6" "checksum atomicwrites 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c4e1aa99513c90202b4b04cfbe3c9d51dd914f2e26215a4caa76574b00bb6393" "checksum atty 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)" = "1803c647a3ec87095e7ae7acfca019e98de5ec9a7d01343f611cf3152ed71a90" -"checksum autocfg 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "22130e92352b948e7e82a49cdb0aa94f2211761117f29e052dd397c1ac33542b" -"checksum backtrace 0.3.34 (registry+https://github.com/rust-lang/crates.io-index)" = "b5164d292487f037ece34ec0de2fcede2faa162f085dd96d2385ab81b12765ba" +"checksum autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "b671c8fb71b457dd4ae18c4ba1e59aa81793daacc361d82fcd410cef0d491875" +"checksum backtrace 0.3.35 (registry+https://github.com/rust-lang/crates.io-index)" = "1371048253fa3bac6704bfd6bbfc922ee9bdcee8881330d40f308b81cc5adc55" "checksum backtrace-sys 0.1.31 (registry+https://github.com/rust-lang/crates.io-index)" = "82a830b4ef2d1124a711c71d263c5abdc710ef8e907bd508c88be475cebc422b" "checksum base64 0.10.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0b25d992356d2eb0ed82172f5248873db5560c4721f564b13cb5193bda5e668e" "checksum bit-set 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e84c238982c4b1e1ee668d136c510c67a13465279c0cb367ea6baf6310620a80" @@ -2550,7 +2549,7 @@ dependencies = [ "checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" "checksum futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)" = "45dc39533a6cae6da2b56da48edae506bb767ec07370f86f70fc062e9d435869" "checksum futures-cpupool 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4" -"checksum getrandom 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)" = "2512b3191f22e2763a5db387f1c9409379772e2050841722eb4a8c4f497bf096" +"checksum getrandom 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)" = "6171a6cc63fbabbe27c2b5ee268e8b7fe5dc1eb0dd2dfad537c1dfed6f69117e" "checksum h2 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)" = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462" "checksum hashbrown 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2bcea5b597dd98e6d1f1ec171744cc5dee1a30d1c23c5b98e3cf9d4fbdf8a526" "checksum heck 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "20564e78d53d2bb135c343b3f47714a56af2061f1c928fdb541dc7b9fdd94205" @@ -2562,7 +2561,7 @@ dependencies = [ "checksum httparse 1.3.4 (registry+https://github.com/rust-lang/crates.io-index)" = "cd179ae861f0c2e53da70d892f5f3029f9594be0c41dc5269cd371691b1dc2f9" "checksum hyper 0.12.33 (registry+https://github.com/rust-lang/crates.io-index)" = "7cb44cbce9d8ee4fb36e4c0ad7b794ac44ebaad924b9c8291a63215bb44c2c8f" "checksum idna 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "02e2673c30ee86b5b96a9cb52ad15718aa1f966f5ab9ad54a8b95d5ca33120a9" -"checksum indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7e81a7c05f79578dbc15793d8b619db9ba32b4577003ef3af1a91c416798c58d" +"checksum indexmap 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "a4d6d89e0948bf10c08b9ecc8ac5b83f07f857ebe2c0cbe38de15b4e4f510356" "checksum inotify 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)" = "40b54539f3910d6f84fbf9a643efd6e3aa6e4f001426c0329576128255994718" "checksum inotify-sys 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "e74a1aa87c59aeff6ef2cc2fa62d41bc43f54952f55652656b18a02fd5e356c0" "checksum iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "dbe6e417e7d0975db6512b90796e8ce223145ac4e33c377e4a42882a0e88bb08" @@ -2624,7 +2623,7 @@ dependencies = [ "checksum prost-derive 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5e7dc378b94ac374644181a2247cebf59a6ec1c88b49ac77f3a94b86b79d0e11" "checksum prost-types 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1de482a366941c8d56d19b650fac09ca08508f2a696119ee7513ad590c8bac6f" "checksum quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)" = "6ce23b6b870e8f94f81fb0a363d65d86675884b34a09043c81e5562f11c1f8e1" -"checksum quote 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "49d77c41ca8767f2f41394c11a4eebccab83da25e7cc035387a3125f02be90a3" +"checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe" "checksum rand 0.3.23 (registry+https://github.com/rust-lang/crates.io-index)" = "64ac302d8f83c0c1974bf758f6b041c6c8ada916fbb44a609158ca8b064cc76c" "checksum rand 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "552840b97013b1a26992c11eac34bdd778e464601a4c2054b5f0bff7c6761293" "checksum rand 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c618c47cd3ebd209790115ab837de41425723956ad3ce2e6a7f09890947cacb9" @@ -2680,7 +2679,7 @@ dependencies = [ "checksum string 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "d24114bfcceb867ca7f71a0d3fe45d45619ec47a6fbfa98cb14e14250bfa5d6d" "checksum strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a" "checksum syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)" = "9ca4b3b69a77cbe1ffc9e198781b7acb0c7365a883670e8f1c1bc66fba79a5c5" -"checksum syn 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "2ae5cd13590144ea968ba5d5520da7a4c08415861014399b5b349f74591c375f" +"checksum syn 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)" = "158521e6f544e7e3dcfc370ac180794aa38cb34a1b1e07609376d4adcf429b93" "checksum synstructure 0.10.2 (registry+https://github.com/rust-lang/crates.io-index)" = "02353edf96d6e4dc81aea2d8490a7e9db177bf8acb0e951c24940bf866cb313f" "checksum tantivy 0.10.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a336e43be781173cff6793242157791dcfc6d3d7a3c92c9fc0a9c8828e0822e5" "checksum tantivy-fst 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "16b22af5ff09b8897093287642a5aaee6f30eb496526ef83a8dd0f4c636ac367" @@ -2721,8 +2720,8 @@ dependencies = [ "checksum tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2cc0c98637d23732f8de6dfd16494c9f1559c3b9e20b4a46462c8f9b9e827bfa" "checksum tower-timeout 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5c06bbc2fbd056f810940a8c6f0cc194557d36da3c22999a755a7a6612447da9" "checksum tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "4792342fac093db5d2558655055a89a04ca909663467a4310c7739d9f8b64698" -"checksum tracing 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "e2b3dd549612b88fd19635628c1e6773f557ac419eece8deaefab1295928324d" -"checksum tracing-attributes 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "af9da7f5256f2d11e322c9f10ba848a3eefae64f0272df5e7100dcda53892801" +"checksum tracing 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c42f6ea7b45c17427e628509f5fac99debf2869760c6248aa3f95fc035313ade" +"checksum tracing-attributes 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "5b43a1835d0cb99f4a36fcdd0f777f72e4d4ff2eb6e78a0e105ac25e41309efa" "checksum tracing-core 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "e94af5e2a5f1700cc58127f93d4b7e46c2b925856592066b9880aabce633b6d8" "checksum tracing-fmt 0.0.1-alpha.3 (registry+https://github.com/rust-lang/crates.io-index)" = "5b7dced65cefbb0445d76d28d023be32527d6b96ef02b2db9b65d1632d293cc7" "checksum tracing-futures 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)" = "08c7446f4fb35df7ba2c537b7e2f812f91b20a58aa2b846f028342c4d2429be0" @@ -2732,7 +2731,7 @@ dependencies = [ "checksum unicode-bidi 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)" = "49f2bd0c6468a8230e1db229cff8029217cf623c767ea5d60bfbd42729ea54d5" "checksum unicode-normalization 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "141339a08b982d942be2ca06ff8b076563cbe223d1befd5450716790d44e2426" "checksum unicode-segmentation 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1967f4cdfc355b37fd76d2a954fb2ed3871034eb4f26d60537d88795cfc332a9" -"checksum unicode-width 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "882386231c45df4700b275c7ff55b6f3698780a650026380e72dabe76fa46526" +"checksum unicode-width 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "7007dbd421b92cc6e28410fe7362e2e0a2503394908f417b68ec8d1c364c4e20" "checksum unicode-xid 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "fc72304796d0818e357ead4e000d19c9c174ab23dc11093ac919054d20a6a7fc" "checksum unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c" "checksum unreachable 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "382810877fe448991dfc7f0dd6e3ae5d58088fd0ea5e35189655f84e6814fa56" diff --git a/src/bin/toshi.rs b/src/bin/toshi.rs index a9e51084..c9102b87 100644 --- a/src/bin/toshi.rs +++ b/src/bin/toshi.rs @@ -77,15 +77,18 @@ pub fn main() -> Result<(), ()> { .wait() } -fn create_watcher(catalog: Arc>, settings: &Settings) -> impl Future { +fn create_watcher(catalog: Arc>, settings: &Settings) -> (Arc, impl Future) { + let commit_watcher = Arc::new(IndexWatcher::new(Arc::clone(&catalog), settings.auto_commit_duration)); if settings.auto_commit_duration > 0 { - let commit_watcher = IndexWatcher::new(catalog.clone(), settings.auto_commit_duration); - future::Either::A(future::lazy(move || { - commit_watcher.start(); - future::ok::<(), ()>(()) - })) + ( + Arc::clone(&commit_watcher), + future::Either::A(future::lazy(move || { + Arc::clone(&commit_watcher).start(); + future::ok::<(), ()>(()) + })), + ) } else { - future::Either::B(future::ok::<(), ()>(())) + (commit_watcher, future::Either::B(future::ok::<(), ()>(()))) } } @@ -100,11 +103,11 @@ fn run_data(catalog: Arc>, settings: &Settings) -> impl Fut println!("{}", RPC_HEADER); info!("I am a data node...Binding to: {}", addr); - commit_watcher.and_then(move |_| RpcServer::serve(bind, catalog)) + commit_watcher.1.and_then(move |_| RpcServer::serve(bind, catalog)) } fn run_master(catalog: Arc>, settings: &Settings) -> impl Future { - let commit_watcher = create_watcher(Arc::clone(&catalog), settings); + let (watcher, commit_watcher) = create_watcher(Arc::clone(&catalog), settings); let addr: IpAddr = settings .host .parse() @@ -135,11 +138,11 @@ fn run_master(catalog: Arc>, settings: &Settings) -> impl F let update = catalog.read().update_remote_indexes(); tokio::spawn(update); } - router_with_catalog(&bind, Arc::clone(&catalog)) + router_with_catalog(&bind, Arc::clone(&catalog), Arc::clone(&watcher)) }); future::Either::A(run) } else { - let run = commit_watcher.and_then(move |_| router_with_catalog(&bind, Arc::clone(&catalog))); + let run = commit_watcher.and_then(move |_| router_with_catalog(&bind, Arc::clone(&catalog), Arc::clone(&watcher))); future::Either::B(run) } } diff --git a/src/commit.rs b/src/commit.rs index 76a7292e..5588e313 100644 --- a/src/commit.rs +++ b/src/commit.rs @@ -1,3 +1,4 @@ +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -6,21 +7,36 @@ use parking_lot::RwLock; use tokio::timer::Interval; use tracing::*; -use crate::index::IndexCatalog; +use crate::index::{IndexCatalog, SharedCatalog}; pub struct IndexWatcher { commit_duration: u64, - catalog: Arc>, + catalog: SharedCatalog, + bulk: Arc, } impl IndexWatcher { - pub fn new(catalog: Arc>, commit_duration: u64) -> Self { - IndexWatcher { catalog, commit_duration } + pub fn new(catalog: SharedCatalog, commit_duration: u64) -> Self { + IndexWatcher { + catalog, + commit_duration, + bulk: Arc::new(AtomicBool::new(false)), + } } - pub fn start(self) { + pub fn bulk_lock(&self) { + self.bulk.store(true, Ordering::SeqCst); + } + + pub fn bulk_unlock(&self) { + self.bulk.store(false, Ordering::SeqCst); + } + + pub fn start(&self) -> impl Future + Send + Sync + 'static { let catalog = Arc::clone(&self.catalog); - let task = Interval::new_interval(Duration::from_secs(self.commit_duration)) + // let cat = catalog.read(); + + Interval::new_interval(Duration::from_secs(self.commit_duration)) .for_each(move |_| { let cat = catalog.read(); cat.get_collection().into_iter().for_each(|(key, index)| { @@ -28,7 +44,7 @@ impl IndexWatcher { let current_ops = index.get_opstamp(); if current_ops == 0 { debug!("No update to index={}, opstamp={}", key, current_ops); - } else { + } else if !self.bulk.load(Ordering::SeqCst) { let mut w = writer.write(); debug!("Committing {}...", key); w.commit().unwrap(); @@ -38,9 +54,7 @@ impl IndexWatcher { Ok(()) }) - .map_err(|e| panic!("Error in commit-watcher={:?}", e)); - - tokio::spawn(task); + .map_err(|e| panic!("Error in commit-watcher={:?}", e)) } } diff --git a/src/handlers/bulk.rs b/src/handlers/bulk.rs index 596d83f2..24f585af 100644 --- a/src/handlers/bulk.rs +++ b/src/handlers/bulk.rs @@ -13,6 +13,7 @@ use tantivy::{Document, IndexWriter}; use tokio::prelude::*; use tracing::*; +use crate::commit::IndexWatcher; use crate::handlers::ResponseFuture; use crate::index::IndexCatalog; use crate::utils::empty_with_code; @@ -20,14 +21,19 @@ use crate::utils::empty_with_code; #[derive(Clone)] pub struct BulkHandler { catalog: Arc>, + watcher: Arc, } impl BulkHandler { - pub fn new(catalog: Arc>) -> Self { - BulkHandler { catalog } + pub fn new(catalog: Arc>, watcher: Arc) -> Self { + BulkHandler { catalog, watcher } } - fn index_documents(index_writer: Arc>, doc_receiver: Receiver) -> impl Future { + fn index_documents( + index_writer: Arc>, + doc_receiver: Receiver, + watcher: Arc, + ) -> impl Future { future::lazy(move || { let start = Instant::now(); for doc in doc_receiver { @@ -35,12 +41,9 @@ impl BulkHandler { w.add_document(doc); } - log::info!("Piping Documents took: {:?}", start.elapsed()); - let commit = Instant::now(); - let mut w = index_writer.write(); - w.commit().unwrap(); - log::info!("Bulk Commit took: {:?}", commit.elapsed()); - Ok(()) + info!("Piping Documents took: {:?}", start.elapsed()); + info!("Unlocking watcher..."); + Ok(watcher.bulk_unlock()) }) } @@ -61,6 +64,7 @@ impl BulkHandler { } pub fn bulk_insert(&self, body: Body, index: String) -> ResponseFuture { + self.watcher.bulk_lock(); let index_lock = self.catalog.read(); let index_handle = index_lock.get_index(&index).unwrap(); let index = index_handle.get_index(); @@ -71,6 +75,7 @@ impl BulkHandler { let num_threads = index_lock.settings.json_parsing_threads; let line_sender_clone = line_sender.clone(); + let watcher_clone = Arc::clone(&self.watcher); let fut = body .fold(Bytes::new(), move |mut buf, line| -> Result { for _ in 0..num_threads { @@ -97,7 +102,8 @@ impl BulkHandler { if !left.is_empty() { line_sender.send(left).expect("Line sender failed #2"); } - tokio::spawn(BulkHandler::index_documents(writer, doc_recv)); + tokio::spawn(BulkHandler::index_documents(writer, doc_recv, watcher_clone)); + future::ok(empty_with_code(StatusCode::CREATED)) }); @@ -112,18 +118,19 @@ mod tests { use tokio::runtime::Runtime; + use crate::error::Error; use crate::handlers::SearchHandler; use crate::index::tests::*; use crate::results::SearchResults; use super::*; - use crate::error::Error; #[test] fn test_bulk_index() -> Result<(), Error> { let mut runtime = Runtime::new().unwrap(); let server = create_test_catalog("test_index"); - let handler = BulkHandler::new(Arc::clone(&server)); + let watcher = Arc::new(IndexWatcher::new(Arc::clone(&server), 1)); + let handler = BulkHandler::new(Arc::clone(&server), Arc::clone(&watcher)); let body = r#" {"test_text": "asdf1234", "test_i64": 123, "test_u64": 321, "test_unindex": "asdf"} {"test_text": "asdf5678", "test_i64": 456, "test_u64": 678, "test_unindex": "asdf"} diff --git a/src/router.rs b/src/router.rs index 6378dddb..a4092c6f 100644 --- a/src/router.rs +++ b/src/router.rs @@ -7,6 +7,7 @@ use parking_lot::RwLock; use serde::Deserialize; use tokio::prelude::*; +use crate::commit::IndexWatcher; use crate::handlers::*; use crate::index::IndexCatalog; use crate::utils::{not_found, parse_path}; @@ -29,11 +30,15 @@ impl QueryOptions { } } -pub fn router_with_catalog(addr: &SocketAddr, catalog: Arc>) -> impl Future + Send { +pub fn router_with_catalog( + addr: &SocketAddr, + catalog: Arc>, + watcher: Arc, +) -> impl Future + Send { let routes = move || { let search_handler = SearchHandler::new(Arc::clone(&catalog)); let index_handler = IndexHandler::new(Arc::clone(&catalog)); - let bulk_handler = BulkHandler::new(Arc::clone(&catalog)); + let bulk_handler = BulkHandler::new(Arc::clone(&catalog), Arc::clone(&watcher)); let summary_cat = Arc::clone(&catalog); service_fn(move |req: Request| { @@ -68,7 +73,7 @@ pub fn router_with_catalog(addr: &SocketAddr, catalog: Arc> (m, [idx]) if m == Method::POST => search_handler.doc_search(body, idx.to_string()), (m, [idx]) if m == Method::PUT => index_handler.add_document(body, idx.to_string()), (m, [idx]) if m == Method::DELETE => index_handler.delete_term(body, idx.to_string()), - (m, [idx]) if m == Method::GET=> { + (m, [idx]) if m == Method::GET => { if idx == &"favicon.ico" { not_found() } else { @@ -101,7 +106,8 @@ pub mod tests { pub static ref TEST_SERVER: TestServer = { let catalog = create_test_catalog("test_index"); let addr = get_localhost(); - let router = router_with_catalog(&addr, Arc::clone(&catalog)); + let watcher = Arc::new(IndexWatcher::new(Arc::clone(&catalog), 100)); + let router = router_with_catalog(&addr, Arc::clone(&catalog), Arc::clone(&watcher)); TestServer::new(router).expect("Can't start test server") }; }