/*-------------------------------------------------------------------------
 *
 * instrument.c
 *  functions for instrumentation of plan execution
 *
 *
 * Copyright (c) 2001-2025, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *   src/backend/executor/instrument.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>

#include "executor/instrument.h"

BufferUsage pgBufferUsage;
static BufferUsage save_pgBufferUsage;
WalUsage pgWalUsage;
static WalUsage save_pgWalUsage;

static void BufferUsageAdd(BufferUsage *dst, const BufferUsage *add);
static void WalUsageAdd(WalUsage *dst, WalUsage *add);


/* Allocate new instrumentation structure(s) */
Instrumentation *
InstrAlloc(int n, int instrument_options, bool async_mode)
{
 Instrumentation *instr;

 /* initialize all fields to zeroes, then modify as needed */
 instr = palloc0(n * sizeof(Instrumentation));
 if (instrument_options & (INSTRUMENT_BUFFERS | INSTRUMENT_TIMER | INSTRUMENT_WAL))
 {
  bool  need_buffers = (instrument_options & INSTRUMENT_BUFFERS) != 0;
  bool  need_wal = (instrument_options & INSTRUMENT_WAL) != 0;
  bool  need_timer = (instrument_options & INSTRUMENT_TIMER) != 0;
  int   i;

  for (i = 0; i < n; i++)
  {
   instr[i].need_bufusage = need_buffers;
   instr[i].need_walusage = need_wal;
   instr[i].need_timer = need_timer;
   instr[i].async_mode = async_mode;
  }
 }

 return instr;
}

/* Initialize a pre-allocated instrumentation structure. */
void
InstrInit(Instrumentation *instr, int instrument_options)
{
 memset(instr, 0, sizeof(Instrumentation));
 instr->need_bufusage = (instrument_options & INSTRUMENT_BUFFERS) != 0;
 instr->need_walusage = (instrument_options & INSTRUMENT_WAL) != 0;
 instr->need_timer = (instrument_options & INSTRUMENT_TIMER) != 0;
}

/* Entry to a plan node */
void
InstrStartNode(Instrumentation *instr)
{
 if (instr->need_timer &&
  !INSTR_TIME_SET_CURRENT_LAZY(instr->starttime))
  elog(ERROR, "InstrStartNode called twice in a row");

 /* save buffer usage totals at node entry, if needed */
 if (instr->need_bufusage)
  instr->bufusage_start = pgBufferUsage;

 if (instr->need_walusage)
  instr->walusage_start = pgWalUsage;
}

/* Exit from a plan node */
void
InstrStopNode(Instrumentation *instr, double nTuples)
{
 double  save_tuplecount = instr->tuplecount;
 instr_time endtime;

 /* count the returned tuples */
 instr->tuplecount += nTuples;

 /* let's update the time only if the timer was requested */
 if (instr->need_timer)
 {
  if (INSTR_TIME_IS_ZERO(instr->starttime))
   elog(ERROR, "InstrStopNode called without start");

  INSTR_TIME_SET_CURRENT(endtime);
  INSTR_TIME_ACCUM_DIFF(instr->counter, endtime, instr->starttime);

  INSTR_TIME_SET_ZERO(instr->starttime);
 }

 /* Add delta of buffer usage since entry to node's totals */
 if (instr->need_bufusage)
  BufferUsageAccumDiff(&instr->bufusage,
        &pgBufferUsage, &instr->bufusage_start);

 if (instr->need_walusage)
  WalUsageAccumDiff(&instr->walusage,
        &pgWalUsage, &instr->walusage_start);

 /* Is this the first tuple of this cycle? */
 if (!instr->running)
 {
  instr->running = true;
  instr->firsttuple = INSTR_TIME_GET_DOUBLE(instr->counter);
 }
 else
 {
  /*
   * In async mode, if the plan node hadn't emitted any tuples before,
   * this might be the first tuple
 */
  if (instr->async_mode && save_tuplecount < 1.0)
   instr->firsttuple = INSTR_TIME_GET_DOUBLE(instr->counter);
 }
}

/* Update tuple count */
void
InstrUpdateTupleCount(Instrumentation *instr, double nTuples)
{
 /* count the returned tuples */
 instr->tuplecount += nTuples;
}

/* Finish a run cycle for a plan node */
void
InstrEndLoop(Instrumentation *instr)
{
 double  totaltime;

 /* Skip if nothing has happened, or already shut down */
 if (!instr->running)
  return;

 if (!INSTR_TIME_IS_ZERO(instr->starttime))
  elog(ERROR, "InstrEndLoop called on running node");

 /* Accumulate per-cycle statistics into totals */
 totaltime = INSTR_TIME_GET_DOUBLE(instr->counter);

 instr->startup += instr->firsttuple;
 instr->total += totaltime;
 instr->ntuples += instr->tuplecount;
 instr->nloops += 1;

 /* Reset for next cycle (if any) */
 instr->running = false;
 INSTR_TIME_SET_ZERO(instr->starttime);
 INSTR_TIME_SET_ZERO(instr->counter);
 instr->firsttuple = 0;
 instr->tuplecount = 0;
}

/* aggregate instrumentation information */
void
InstrAggNode(Instrumentation *dst, Instrumentation *add)
{
 if (!dst->running && add->running)
 {
  dst->running = true;
  dst->firsttuple = add->firsttuple;
 }
 else if (dst->running && add->running && dst->firsttuple > add->firsttuple)
  dst->firsttuple = add->firsttuple;

 INSTR_TIME_ADD(dst->counter, add->counter);

 dst->tuplecount += add->tuplecount;
 dst->startup += add->startup;
 dst->total += add->total;
 dst->ntuples += add->ntuples;
 dst->ntuples2 += add->ntuples2;
 dst->nloops += add->nloops;
 dst->nfiltered1 += add->nfiltered1;
 dst->nfiltered2 += add->nfiltered2;

 /* Add delta of buffer usage since entry to node's totals */
 if (dst->need_bufusage)
  BufferUsageAdd(&dst->bufusage, &add->bufusage);

 if (dst->need_walusage)
  WalUsageAdd(&dst->walusage, &add->walusage);
}

/* note current values during parallel executor startup */
void
InstrStartParallelQuery(void)
{
 save_pgBufferUsage = pgBufferUsage;
 save_pgWalUsage = pgWalUsage;
}

/* report usage after parallel executor shutdown */
void
InstrEndParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
{
 memset(bufusage, 0, sizeof(BufferUsage));
 BufferUsageAccumDiff(bufusage, &pgBufferUsage, &save_pgBufferUsage);
 memset(walusage, 0, sizeof(WalUsage));
 WalUsageAccumDiff(walusage, &pgWalUsage, &save_pgWalUsage);
}

/* accumulate work done by workers in leader's stats */
void
InstrAccumParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
{
 BufferUsageAdd(&pgBufferUsage, bufusage);
 WalUsageAdd(&pgWalUsage, walusage);
}

/* dst += add */
static void
BufferUsageAdd(BufferUsage *dst, const BufferUsage *add)
{
 dst->shared_blks_hit += add->shared_blks_hit;
 dst->shared_blks_read += add->shared_blks_read;
 dst->shared_blks_dirtied += add->shared_blks_dirtied;
 dst->shared_blks_written += add->shared_blks_written;
 dst->local_blks_hit += add->local_blks_hit;
 dst->local_blks_read += add->local_blks_read;
 dst->local_blks_dirtied += add->local_blks_dirtied;
 dst->local_blks_written += add->local_blks_written;
 dst->temp_blks_read += add->temp_blks_read;
 dst->temp_blks_written += add->temp_blks_written;
 INSTR_TIME_ADD(dst->shared_blk_read_time, add->shared_blk_read_time);
 INSTR_TIME_ADD(dst->shared_blk_write_time, add->shared_blk_write_time);
 INSTR_TIME_ADD(dst->local_blk_read_time, add->local_blk_read_time);
 INSTR_TIME_ADD(dst->local_blk_write_time, add->local_blk_write_time);
 INSTR_TIME_ADD(dst->temp_blk_read_time, add->temp_blk_read_time);
 INSTR_TIME_ADD(dst->temp_blk_write_time, add->temp_blk_write_time);
}

/* dst += add - sub */
void
BufferUsageAccumDiff(BufferUsage *dst,
      const BufferUsage *add,
      const BufferUsage *sub)
{
 dst->shared_blks_hit += add->shared_blks_hit - sub->shared_blks_hit;
 dst->shared_blks_read += add->shared_blks_read - sub->shared_blks_read;
 dst->shared_blks_dirtied += add->shared_blks_dirtied - sub->shared_blks_dirtied;
 dst->shared_blks_written += add->shared_blks_written - sub->shared_blks_written;
 dst->local_blks_hit += add->local_blks_hit - sub->local_blks_hit;
 dst->local_blks_read += add->local_blks_read - sub->local_blks_read;
 dst->local_blks_dirtied += add->local_blks_dirtied - sub->local_blks_dirtied;
 dst->local_blks_written += add->local_blks_written - sub->local_blks_written;
 dst->temp_blks_read += add->temp_blks_read - sub->temp_blks_read;
 dst->temp_blks_written += add->temp_blks_written - sub->temp_blks_written;
 INSTR_TIME_ACCUM_DIFF(dst->shared_blk_read_time,
        add->shared_blk_read_time, sub->shared_blk_read_time);
 INSTR_TIME_ACCUM_DIFF(dst->shared_blk_write_time,
        add->shared_blk_write_time, sub->shared_blk_write_time);
 INSTR_TIME_ACCUM_DIFF(dst->local_blk_read_time,
        add->local_blk_read_time, sub->local_blk_read_time);
 INSTR_TIME_ACCUM_DIFF(dst->local_blk_write_time,
        add->local_blk_write_time, sub->local_blk_write_time);
 INSTR_TIME_ACCUM_DIFF(dst->temp_blk_read_time,
        add->temp_blk_read_time, sub->temp_blk_read_time);
 INSTR_TIME_ACCUM_DIFF(dst->temp_blk_write_time,
        add->temp_blk_write_time, sub->temp_blk_write_time);
}

/* helper functions for WAL usage accumulation */
static void
WalUsageAdd(WalUsage *dst, WalUsage *add)
{
 dst->wal_bytes += add->wal_bytes;
 dst->wal_records += add->wal_records;
 dst->wal_fpi += add->wal_fpi;
 dst->wal_buffers_full += add->wal_buffers_full;
}

void
WalUsageAccumDiff(WalUsage *dst, const WalUsage *add, const WalUsage *sub)
{
 dst->wal_bytes += add->wal_bytes - sub->wal_bytes;
 dst->wal_records += add->wal_records - sub->wal_records;
 dst->wal_fpi += add->wal_fpi - sub->wal_fpi;
 dst->wal_buffers_full += add->wal_buffers_full - sub->wal_buffers_full;
}