| 25 | #include "flow/actorcompiler.h" // This must be the last #include. |
| 26 | |
| 27 | struct GetRangeStream : TestWorkload { |
| 28 | PerfIntCounter bytesRead; |
| 29 | bool useGetRange; |
| 30 | Key begin; |
| 31 | Key end; |
| 32 | bool printKVPairs; |
| 33 | |
| 34 | GetRangeStream(WorkloadContext const& wcx) : TestWorkload(wcx), bytesRead("BytesRead") { |
| 35 | useGetRange = getOption(options, LiteralStringRef("useGetRange"), false); |
| 36 | begin = getOption(options, LiteralStringRef("begin"), normalKeys.begin); |
| 37 | end = getOption(options, LiteralStringRef("end"), normalKeys.end); |
| 38 | printKVPairs = getOption(options, LiteralStringRef("printKVPairs"), false); |
| 39 | } |
| 40 | |
| 41 | std::string description() const override { return "GetRangeStreamWorkload"; } |
| 42 | |
| 43 | Future<Void> setup(Database const& cx) override { return Void(); } |
| 44 | |
| 45 | Future<Void> start(Database const& cx) override { |
| 46 | return clientId != 0 ? Void() : useGetRange ? fdbClientGetRange(cx, this) : fdbClientStream(cx, this); |
| 47 | } |
| 48 | |
| 49 | Future<bool> check(Database const& cx) override { return true; } |
| 50 | |
| 51 | void getMetrics(std::vector<PerfMetric>& m) override { m.push_back(bytesRead.getMetric()); } |
| 52 | |
| 53 | ACTOR static Future<Void> logThroughput(GetRangeStream* self, Key* next) { |
| 54 | loop { |
| 55 | state int64_t last = self->bytesRead.getValue(); |
| 56 | state double before = g_network->now(); |
| 57 | wait(delay(1)); |
| 58 | state double after = g_network->now(); |
| 59 | if (after > before) { |
| 60 | printf("throughput: %g bytes/s, next: %s\n", |
| 61 | (self->bytesRead.getValue() - last) / (after - before), |
| 62 | printable(*next).c_str()); |
| 63 | } |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | ACTOR static Future<Void> fdbClientGetRange(Database db, GetRangeStream* self) { |
| 68 | state Transaction tx(db); |
| 69 | state Key next = self->begin; |
| 70 | state Future<Void> logFuture = logThroughput(self, &next); |
| 71 | loop { |
| 72 | try { |
| 73 | Standalone<RangeResultRef> range = wait( |
| 74 | tx.getRange(KeySelector(firstGreaterOrEqual(next), next.arena()), |
| 75 | KeySelector(firstGreaterOrEqual(self->end)), |
| 76 | GetRangeLimits(GetRangeLimits::ROW_LIMIT_UNLIMITED, CLIENT_KNOBS->REPLY_BYTE_LIMIT))); |
| 77 | for (const auto& [k, v] : range) { |
| 78 | if (self->printKVPairs) { |
| 79 | printf("%s -> %s\n", printable(k).c_str(), printable(v).c_str()); |
| 80 | } |
| 81 | self->bytesRead += k.size() + v.size(); |
| 82 | } |
| 83 | if (!range.more) { |
| 84 | break; |
no test coverage detected