tqueue.c File Reference

#include "postgres.h"
#include "access/htup_details.h"
#include "executor/tqueue.h"

Include dependency graph for tqueue.c:

Go to the source code of this file.

Data Structures
struct	TQueueDestReceiver
struct	TupleQueueReader

Typedefs
typedef struct TQueueDestReceiver	TQueueDestReceiver

Functions
static bool	tqueueReceiveSlot (TupleTableSlot *slot, DestReceiver *self)
static void	tqueueStartupReceiver (DestReceiver *self, int operation, TupleDesc typeinfo)
static void	tqueueShutdownReceiver (DestReceiver *self)
static void	tqueueDestroyReceiver (DestReceiver *self)
DestReceiver *	CreateTupleQueueDestReceiver (shm_mq_handle *handle)
TupleQueueReader *	CreateTupleQueueReader (shm_mq_handle *handle)
void	DestroyTupleQueueReader (TupleQueueReader *reader)
MinimalTuple	TupleQueueReaderNext (TupleQueueReader *reader, bool nowait, bool *done)

Typedef Documentation

TQueueDestReceiver

typedef struct TQueueDestReceiver TQueueDestReceiver

Function Documentation

CreateTupleQueueDestReceiver()

DestReceiver * CreateTupleQueueDestReceiver

(

shm_mq_handle *

handle

)

Definition at line 119 of file tqueue.c.

{
    TQueueDestReceiver *self;
 
    self = (TQueueDestReceiver *) palloc0(sizeof(TQueueDestReceiver));
 
    self->pub.receiveSlot = tqueueReceiveSlot;
    self->pub.rStartup = tqueueStartupReceiver;
    self->pub.rShutdown = tqueueShutdownReceiver;
    self->pub.rDestroy = tqueueDestroyReceiver;
    self->pub.mydest = DestTupleQueue;
    self->queue = handle;
 
    return (DestReceiver *) self;
}

References DestTupleQueue, _DestReceiver::mydest, palloc0(), TQueueDestReceiver::pub, TQueueDestReceiver::queue, _DestReceiver::rDestroy, _DestReceiver::receiveSlot, _DestReceiver::rShutdown, _DestReceiver::rStartup, tqueueDestroyReceiver(), tqueueReceiveSlot(), tqueueShutdownReceiver(), and tqueueStartupReceiver().

Referenced by CreateDestReceiver(), and ExecParallelGetReceiver().

CreateTupleQueueReader()

TupleQueueReader * CreateTupleQueueReader

(

shm_mq_handle *

handle

)

Definition at line 139 of file tqueue.c.

{
    TupleQueueReader *reader = palloc0(sizeof(TupleQueueReader));
 
    reader->queue = handle;
 
    return reader;
}

References palloc0(), and TupleQueueReader::queue.

Referenced by ExecParallelCreateReaders().

DestroyTupleQueueReader()

void DestroyTupleQueueReader

(

TupleQueueReader *

reader

)

Definition at line 155 of file tqueue.c.

{
    pfree(reader);
}

References pfree().

Referenced by ExecParallelFinish().

tqueueDestroyReceiver()

static void tqueueDestroyReceiver ( DestReceiver *  self )

static

Definition at line 105 of file tqueue.c.

{
    TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
 
    /* We probably already detached from queue, but let's be sure */
    if (tqueue->queue != NULL)
        shm_mq_detach(tqueue->queue);
    pfree(self);
}

References pfree(), TQueueDestReceiver::queue, and shm_mq_detach().

Referenced by CreateTupleQueueDestReceiver().

tqueueReceiveSlot()

static bool tqueueReceiveSlot ( TupleTableSlot *  slot, DestReceiver *  self  )

static

Definition at line 54 of file tqueue.c.

{
    TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
    MinimalTuple tuple;
    shm_mq_result result;
    bool        should_free;
 
    /* Send the tuple itself. */
    tuple = ExecFetchSlotMinimalTuple(slot, &should_free);
    result = shm_mq_send(tqueue->queue, tuple->t_len, tuple, false, false);
 
    if (should_free)
        pfree(tuple);
 
    /* Check for failure. */
    if (result == SHM_MQ_DETACHED)
        return false;
    else if (result != SHM_MQ_SUCCESS)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("could not send tuple to shared-memory queue")));
 
    return true;
}

References ereport, errcode(), errmsg(), ERROR, ExecFetchSlotMinimalTuple(), pfree(), TQueueDestReceiver::queue, SHM_MQ_DETACHED, shm_mq_send(), SHM_MQ_SUCCESS, and MinimalTupleData::t_len.

Referenced by CreateTupleQueueDestReceiver().

tqueueShutdownReceiver()

static void tqueueShutdownReceiver ( DestReceiver *  self )

static

Definition at line 92 of file tqueue.c.

{
    TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
 
    if (tqueue->queue != NULL)
        shm_mq_detach(tqueue->queue);
    tqueue->queue = NULL;
}

References TQueueDestReceiver::queue, and shm_mq_detach().

Referenced by CreateTupleQueueDestReceiver().

tqueueStartupReceiver()

static void tqueueStartupReceiver ( DestReceiver *  self, int  operation, TupleDesc  typeinfo  )

static

Definition at line 83 of file tqueue.c.

{
    /* do nothing */
}

Referenced by CreateTupleQueueDestReceiver().

TupleQueueReaderNext()

MinimalTuple TupleQueueReaderNext	(	TupleQueueReader *	reader,
		bool	nowait,
		bool *	done
	)

Definition at line 176 of file tqueue.c.

{
    MinimalTuple tuple;
    shm_mq_result result;
    Size        nbytes;
    void       *data;
 
    if (done != NULL)
        *done = false;
 
    /* Attempt to read a message. */
    result = shm_mq_receive(reader->queue, &nbytes, &data, nowait);
 
    /* If queue is detached, set *done and return NULL. */
    if (result == SHM_MQ_DETACHED)
    {
        if (done != NULL)
            *done = true;
        return NULL;
    }
 
    /* In non-blocking mode, bail out if no message ready yet. */
    if (result == SHM_MQ_WOULD_BLOCK)
        return NULL;
    Assert(result == SHM_MQ_SUCCESS);
 
    /*
     * Return a pointer to the queue memory directly (which had better be
     * sufficiently aligned).
     */
    tuple = (MinimalTuple) data;
    Assert(tuple->t_len == nbytes);
 
    return tuple;
}

References Assert, data, TupleQueueReader::queue, SHM_MQ_DETACHED, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, and MinimalTupleData::t_len.

Referenced by gather_readnext(), and gm_readnext_tuple().