Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit04dc7fe

Browse files
committed
Patch for stable execute at 3-node cluster
1 parentf84be93 commit04dc7fe

File tree

12 files changed

+291
-269
lines changed

12 files changed

+291
-269
lines changed

‎contrib/pg_exchange/Makefile

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@ EXTENSION = pg_exchange
55
EXTVERSION = 0.1
66
PGFILEDESC = "pg_exchange - an exchange custom node and rules for the planner"
77
MODULES = pg_exchange
8-
OBJS = pg_exchange.o exchange.o hooks.o common.o nodeDummyscan.o nodeDistPlanExec.o dmq.o stream.o partutils.o$(WIN32RES)
8+
OBJS =common.o dmq.o exchange.o expath.o hooks.o nodeDistPlanExec.o\
9+
nodeDummyscan.o partutils.o pg_exchange.o stream.o$(WIN32RES)
910

1011
fdw_srcdir =$(top_srcdir)/contrib/postgres_fdw/
1112

‎contrib/pg_exchange/dmq.c

Lines changed: 40 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1239,27 +1239,44 @@ dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg)
12391239
resetStringInfo(&buf);
12401240
}
12411241

1242+
staticboolpush_state= false;
1243+
staticStringInfoDatabuf;
12421244

1243-
void
1244-
dmq_push_buffer(DmqDestinationIddest_id,char*stream_name,constvoid*payload,size_tlen)
1245+
bool
1246+
dmq_push_buffer(DmqDestinationIddest_id,char*stream_name,
1247+
constvoid*payload,size_tlen,boolnowait)
12451248
{
1246-
StringInfoDatabuf;
12471249
shm_mq_resultres;
12481250

1249-
ensure_outq_handle();
1251+
if (!push_state)
1252+
{
1253+
ensure_outq_handle();
12501254

1251-
initStringInfo(&buf);
1252-
pq_sendbyte(&buf,dest_id);
1253-
pq_send_ascii_string(&buf,stream_name);
1254-
pq_sendbytes(&buf,payload,len);
1255+
initStringInfo(&buf);
1256+
pq_sendbyte(&buf,dest_id);
1257+
pq_send_ascii_string(&buf,stream_name);
1258+
pq_sendbytes(&buf,payload,len);
12551259

1256-
mtm_log(DmqTraceOutgoing,"[DMQ] pushing l=%d '%.*s'",
1257-
buf.len,buf.len,buf.data);
1260+
mtm_log(DmqTraceOutgoing,"[DMQ] pushing l=%d '%.*s'",
1261+
buf.len,buf.len,buf.data);
1262+
}
12581263

12591264
// XXX: use sendv instead
1260-
res=shm_mq_send(dmq_local.mq_outh,buf.len,buf.data, false);
1265+
res=shm_mq_send(dmq_local.mq_outh,buf.len,buf.data,nowait);
1266+
1267+
if (res==SHM_MQ_WOULD_BLOCK)
1268+
{
1269+
Assert(nowait== true);
1270+
push_state= true;
1271+
/* Report on full queue. */
1272+
return false;
1273+
}
1274+
12611275
if (res!=SHM_MQ_SUCCESS)
12621276
mtm_log(WARNING,"[DMQ] dmq_push: can't send to queue");
1277+
1278+
push_state= false;
1279+
return true;
12631280
}
12641281

12651282
staticbool
@@ -1467,6 +1484,17 @@ dmq_sender_name(DmqSenderId id)
14671484
returndmq_local.inhandles[id].name;
14681485
}
14691486

1487+
char*
1488+
dmq_receiver_name(DmqDestinationIddest_id)
1489+
{
1490+
char*recvName;
1491+
1492+
LWLockAcquire(dmq_state->lock,LW_SHARED);
1493+
recvName=pstrdup(dmq_state->destinations[dest_id].receiver_name);
1494+
LWLockRelease(dmq_state->lock);
1495+
returnrecvName;
1496+
}
1497+
14701498
DmqDestinationId
14711499
dmq_remote_id(constchar*name)
14721500
{
@@ -1492,7 +1520,7 @@ dmq_remote_id(const char *name)
14921520
* received. Returns false, if an error is occured.
14931521
*
14941522
* sender_id - identifier of the received message sender.
1495-
* msg - buffer that contains received message.
1523+
* msg -pointer to localbuffer that contains received message.
14961524
* len - size of received message.
14971525
*/
14981526
constchar*

‎contrib/pg_exchange/dmq.h

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ typedef enum
1515
typedefint8DmqDestinationId;
1616
typedefint8DmqSenderId;
1717

18-
#defineDMQ_NAME_MAXLEN32
18+
#defineDMQ_NAME_MAXLEN64
1919
#defineDMQ_MAX_DESTINATIONS 127
2020
#defineDMQ_MAX_RECEIVERS 100
2121

@@ -38,6 +38,7 @@ extern void dmq_terminate_receiver(char *name);
3838
externvoiddmq_stream_subscribe(constchar*stream_name);
3939
externvoiddmq_stream_unsubscribe(constchar*stream_name);
4040
externconstchar*dmq_sender_name(DmqSenderIdid);
41+
externchar*dmq_receiver_name(DmqDestinationIddest_id);
4142
externDmqDestinationIddmq_remote_id(constchar*name);
4243

4344
externconstchar*
@@ -46,8 +47,8 @@ dmq_pop(DmqSenderId *sender_id, void **msg, Size *len, uint64 mask,
4647
externbooldmq_pop_nb(DmqSenderId*sender_id,StringInfomsg,uint64mask);
4748

4849
externvoiddmq_push(DmqDestinationIddest_id,char*stream_name,char*msg);
49-
externvoiddmq_push_buffer(DmqDestinationIddest_id,char*stream_name,
50-
constvoid*buffer,size_tlen);
50+
externbooldmq_push_buffer(DmqDestinationIddest_id,char*stream_name,
51+
constvoid*buffer,size_tlen,boolnowait);
5152

5253
typedefvoid (*dmq_receiver_hook_type) (constchar*);
5354
externdmq_receiver_hook_typedmq_receiver_start_hook;

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp