PostgreSQL Source Code git master
Functions | Variables
test.c File Reference
#include "postgres.h"
#include "fmgr.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "varatt.h"
#include "test_shm_mq.h"
Include dependency graph for test.c:

Go to the source code of this file.

Functions

  PG_FUNCTION_INFO_V1 (test_shm_mq)
 
  PG_FUNCTION_INFO_V1 (test_shm_mq_pipelined)
 
static void  verify_message (Size origlen, char *origdata, Size newlen, char *newdata)
 
Datum  test_shm_mq (PG_FUNCTION_ARGS)
 
Datum  test_shm_mq_pipelined (PG_FUNCTION_ARGS)
 

Variables

  PG_MODULE_MAGIC
 
static uint32  we_message_queue = 0
 

Function Documentation

◆ PG_FUNCTION_INFO_V1() [1/2]

PG_FUNCTION_INFO_V1 ( test_shm_mq  )

◆ PG_FUNCTION_INFO_V1() [2/2]

PG_FUNCTION_INFO_V1 ( test_shm_mq_pipelined  )

◆ test_shm_mq()

Datum test_shm_mq ( PG_FUNCTION_ARGS  )

Definition at line 43 of file test.c.

44{
45 int64 queue_size = PG_GETARG_INT64(0);
46 text *message = PG_GETARG_TEXT_PP(1);
47 char *message_contents = VARDATA_ANY(message);
48 int message_size = VARSIZE_ANY_EXHDR(message);
49 int32 loop_count = PG_GETARG_INT32(2);
50 int32 nworkers = PG_GETARG_INT32(3);
51 dsm_segment *seg;
52 shm_mq_handle *outqh;
53 shm_mq_handle *inqh;
54 shm_mq_result res;
55 Size len;
56 void *data;
57
58 /* A negative loopcount is nonsensical. */
59 if (loop_count < 0)
61 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
62 errmsg("repeat count size must be an integer value greater than or equal to zero")));
63
64 /*
65 * Since this test sends data using the blocking interfaces, it cannot
66 * send data to itself. Therefore, a minimum of 1 worker is required. Of
67 * course, a negative worker count is nonsensical.
68 */
69 if (nworkers <= 0)
71 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
72 errmsg("number of workers must be an integer value greater than zero")));
73
74 /* Set up dynamic shared memory segment and background workers. */
75 test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
76
77 /* Send the initial message. */
78 res = shm_mq_send(outqh, message_size, message_contents, false, true);
79 if (res != SHM_MQ_SUCCESS)
81 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
82 errmsg("could not send message")));
83
84 /*
85 * Receive a message and send it back out again. Do this a number of
86 * times equal to the loop count.
87 */
88 for (;;)
89 {
90 /* Receive a message. */
91 res = shm_mq_receive(inqh, &len, &data, false);
92 if (res != SHM_MQ_SUCCESS)
94 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
95 errmsg("could not receive message")));
96
97 /* If this is supposed to be the last iteration, stop here. */
98 if (--loop_count <= 0)
99 break;
100
101 /* Send it back out. */
102 res = shm_mq_send(outqh, len, data, false, true);
103 if (res != SHM_MQ_SUCCESS)
105 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
106 errmsg("could not send message")));
107 }
108
109 /*
110 * Finally, check that we got back the same message from the last
111 * iteration that we originally sent.
112 */
113 verify_message(message_size, message_contents, len, data);
114
115 /* Clean up. */
116 dsm_detach(seg);
117
119}
int64_t int64
Definition: c.h:535
int32_t int32
Definition: c.h:534
size_t Size
Definition: c.h:610
void dsm_detach(dsm_segment *seg)
Definition: dsm.c:803
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:150
#define PG_RETURN_VOID()
Definition: fmgr.h:349
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:309
#define PG_GETARG_INT64(n)
Definition: fmgr.h:283
#define PG_GETARG_INT32(n)
Definition: fmgr.h:269
const void size_t len
const void * data
void test_shm_mq_setup(int64 queue_size, int32 nworkers, dsm_segment **segp, shm_mq_handle **output, shm_mq_handle **input)
Definition: setup.c:50
shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
Definition: shm_mq.c:572
shm_mq_result shm_mq_send(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, bool force_flush)
Definition: shm_mq.c:329
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
Definition: dsm.c:67
Definition: shm_mq.c:138
Definition: c.h:692
static void verify_message(Size origlen, char *origdata, Size newlen, char *newdata)
Definition: test.c:258
static Size VARSIZE_ANY_EXHDR(const void *PTR)
Definition: varatt.h:472
static char * VARDATA_ANY(const void *PTR)
Definition: varatt.h:486

References data, dsm_detach(), ereport, errcode(), errmsg(), ERROR, len, PG_GETARG_INT32, PG_GETARG_INT64, PG_GETARG_TEXT_PP, PG_RETURN_VOID, shm_mq_receive(), shm_mq_send(), SHM_MQ_SUCCESS, test_shm_mq_setup(), VARDATA_ANY(), VARSIZE_ANY_EXHDR(), and verify_message().

◆ test_shm_mq_pipelined()

Datum test_shm_mq_pipelined ( PG_FUNCTION_ARGS  )

Definition at line 132 of file test.c.

133{
134 int64 queue_size = PG_GETARG_INT64(0);
135 text *message = PG_GETARG_TEXT_PP(1);
136 char *message_contents = VARDATA_ANY(message);
137 int message_size = VARSIZE_ANY_EXHDR(message);
138 int32 loop_count = PG_GETARG_INT32(2);
139 int32 nworkers = PG_GETARG_INT32(3);
140 bool verify = PG_GETARG_BOOL(4);
141 int32 send_count = 0;
142 int32 receive_count = 0;
143 dsm_segment *seg;
144 shm_mq_handle *outqh;
145 shm_mq_handle *inqh;
146 shm_mq_result res;
147 Size len;
148 void *data;
149
150 /* A negative loopcount is nonsensical. */
151 if (loop_count < 0)
153 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
154 errmsg("repeat count size must be an integer value greater than or equal to zero")));
155
156 /*
157 * Using the nonblocking interfaces, we can even send data to ourselves,
158 * so the minimum number of workers for this test is zero.
159 */
160 if (nworkers < 0)
162 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
163 errmsg("number of workers must be an integer value greater than or equal to zero")));
164
165 /* Set up dynamic shared memory segment and background workers. */
166 test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
167
168 /* Main loop. */
169 for (;;)
170 {
171 bool wait = true;
172
173 /*
174 * If we haven't yet sent the message the requisite number of times,
175 * try again to send it now. Note that when shm_mq_send() returns
176 * SHM_MQ_WOULD_BLOCK, the next call to that function must pass the
177 * same message size and contents; that's not an issue here because
178 * we're sending the same message every time.
179 */
180 if (send_count < loop_count)
181 {
182 res = shm_mq_send(outqh, message_size, message_contents, true,
183 true);
184 if (res == SHM_MQ_SUCCESS)
185 {
186 ++send_count;
187 wait = false;
188 }
189 else if (res == SHM_MQ_DETACHED)
191 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
192 errmsg("could not send message")));
193 }
194
195 /*
196 * If we haven't yet received the message the requisite number of
197 * times, try to receive it again now.
198 */
199 if (receive_count < loop_count)
200 {
201 res = shm_mq_receive(inqh, &len, &data, true);
202 if (res == SHM_MQ_SUCCESS)
203 {
204 ++receive_count;
205 /* Verifying every time is slow, so it's optional. */
206 if (verify)
207 verify_message(message_size, message_contents, len, data);
208 wait = false;
209 }
210 else if (res == SHM_MQ_DETACHED)
212 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
213 errmsg("could not receive message")));
214 }
215 else
216 {
217 /*
218 * Otherwise, we've received the message enough times. This
219 * shouldn't happen unless we've also sent it enough times.
220 */
221 if (send_count != receive_count)
223 (errcode(ERRCODE_INTERNAL_ERROR),
224 errmsg("message sent %d times, but received %d times",
225 send_count, receive_count)));
226 break;
227 }
228
229 if (wait)
230 {
231 /* first time, allocate or get the custom wait event */
232 if (we_message_queue == 0)
233 we_message_queue = WaitEventExtensionNew("TestShmMqMessageQueue");
234
235 /*
236 * If we made no progress, wait for one of the other processes to
237 * which we are connected to set our latch, indicating that they
238 * have read or written data and therefore there may now be work
239 * for us to do.
240 */
245 }
246 }
247
248 /* Clean up. */
249 dsm_detach(seg);
250
252}
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
struct Latch * MyLatch
Definition: globals.c:63
void ResetLatch(Latch *latch)
Definition: latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40
static uint32 we_message_queue
Definition: test.c:32
uint32 WaitEventExtensionNew(const char *wait_event_name)
Definition: wait_event.c:163
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34

References CHECK_FOR_INTERRUPTS, data, dsm_detach(), ereport, errcode(), errmsg(), ERROR, len, MyLatch, PG_GETARG_BOOL, PG_GETARG_INT32, PG_GETARG_INT64, PG_GETARG_TEXT_PP, PG_RETURN_VOID, ResetLatch(), SHM_MQ_DETACHED, shm_mq_receive(), shm_mq_send(), SHM_MQ_SUCCESS, test_shm_mq_setup(), VARDATA_ANY(), VARSIZE_ANY_EXHDR(), verify_message(), WaitEventExtensionNew(), WaitLatch(), we_message_queue, WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

◆ verify_message()

static void verify_message ( Size  origlen,
char *  origdata,
Size  newlen,
char *  newdata 
)
static

Definition at line 258 of file test.c.

259{
260 Size i;
261
262 if (origlen != newlen)
264 (errmsg("message corrupted"),
265 errdetail("The original message was %zu bytes but the final message is %zu bytes.",
266 origlen, newlen)));
267
268 for (i = 0; i < origlen; ++i)
269 if (origdata[i] != newdata[i])
271 (errmsg("message corrupted"),
272 errdetail("The new and original messages differ at byte %zu of %zu.", i, origlen)));
273}
int errdetail(const char *fmt,...)
Definition: elog.c:1207
i
int i
Definition: isn.c:77

References ereport, errdetail(), errmsg(), ERROR, and i.

Referenced by test_shm_mq(), and test_shm_mq_pipelined().

Variable Documentation

◆ PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 23 of file test.c.

◆ we_message_queue

uint32 we_message_queue = 0
static

Definition at line 32 of file test.c.

Referenced by test_shm_mq_pipelined().

AltStyle によって変換されたページ (->オリジナル) /