Function CurrentOpWorkerCore

pg_documentdb/src/commands/current_op.c:409–458

* Core data gathering logic in the worker nodes (executes on all workers + coordinator) */

 * Core data gathering logic in the worker nodes (executes on all workers + coordinator)
 */
static pgbson *
CurrentOpWorkerCore(void *specPointer)
{
	pgbson *spec = (pgbson *) specPointer;
	CurrentOpOptions options = { 0 };
	PopulateCurrentOpOptions(spec, &options);

	List *activities = WorkerGetBaseActivities();

	pgbson_writer writer;
	PgbsonWriterInit(&writer);

	pgbson_array_writer childWriter;
	PgbsonWriterStartArray(&writer, "activities", 10, &childWriter);

	ListCell *cell;
	foreach(cell, activities)
	{
		SingleWorkerActivity *activity = lfirst(cell);

		if (!options.idleConnections && !options.idleSessions &&
			strcmp(activity->state, "active") != 0)
		{
			/* Skip inactive sessions if not requested */
			continue;
		}

		if (activity->query != NULL &&
			IsVersionRefreshQueryString(activity->query))
		{
			/* Skip version refresh query calls */
			continue;
		}

		/* by default we would want to just send the activity up - but we really
		 * need to post-process the activity here. This is because things like
		 * index progress need to be handled fully on the worker (the tables aren't
		 * distributed).
		 */
		pgbson_writer singleActivityWriter;
		PgbsonArrayWriterStartDocument(&childWriter, &singleActivityWriter);
		WriteOneActivityToDocument(activity, &singleActivityWriter);
		PgbsonArrayWriterEndDocument(&childWriter, &singleActivityWriter);
	}

	list_free_deep(activities);

	PgbsonWriterEndArray(&writer, &childWriter);
	return PgbsonWriterGetPgbson(&writer);
}


/*