Function SlowLog_Replay

src/slow_log/slow_log.c:264–312

}

void SlowLog_Replay
(
	const SlowLog *slowlog,
	RedisModuleCtx *ctx
) {
	int my_t_id = ThreadPools_GetThreadID();
	SlowLog *aggregated_slowlog = SlowLog_New();

	for(int t_id = 0; t_id < slowlog->count; t_id++) {
		// don't lock ourselves
		if(my_t_id != t_id) {
			if(pthread_mutex_lock(slowlog->locks + t_id) != 0) {
				// failed to lock, skip aggregating this thread slowlog entries
				continue;
			}
		}
		{
			// critical section
			rax *lookup = slowlog->lookup[t_id];
			raxIterator iter;
			raxStart(&iter, lookup);
			raxSeek(&iter, "^", NULL, 0);
			while(raxNext(&iter)) {
				SlowLogItem *item = iter.data;
				SlowLog_Add(aggregated_slowlog, item->cmd, item->query,
							item->latency, &item->time);
			}
			raxStop(&iter);
			// end of critical section
		}
		if(my_t_id != t_id) {
			pthread_mutex_unlock(slowlog->locks + t_id);
		}
	}

	heap_t *heap = aggregated_slowlog->min_heap[my_t_id];
	RedisModule_ReplyWithArray(ctx, Heap_count(heap));

	while(Heap_count(heap)) {
		SlowLogItem *item = Heap_poll(heap);
		RedisModule_ReplyWithArray(ctx, 4);
		RedisModule_ReplyWithDouble(ctx, item->time);
		RedisModule_ReplyWithStringBuffer(ctx, (const char *)item->cmd, strlen(item->cmd));
		RedisModule_ReplyWithStringBuffer(ctx, (const char *)item->query, strlen(item->query));
		_ReplyWithRoundedDouble(ctx, item->latency);
	}

	SlowLog_Free(aggregated_slowlog);
}

void SlowLog_Free(SlowLog *slowlog) {
	for(int i = 0; i < slowlog->count; i++) {