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

Commit0a5f57d

Browse files
committed
Bugfix on restart node problem.
OnNodeDisconnect callback added. DMQ receive error about instance disconnection and call the routine.It is removes entry from hash table ExchShmem->htab referenced to disconnected node, make detach_receiver()and destination_drop().We do not needed in receiver_start callback because check (and establish, if needed) connections beforeeach query.
1 parent47198b5 commit0a5f57d

File tree

6 files changed

+60
-7
lines changed

6 files changed

+60
-7
lines changed

‎contrib/pg_exchange/dmq.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1685,7 +1685,7 @@ dmq_get_destination_status(DmqDestinationId dest_id)
16851685
}
16861686

16871687
void
1688-
dmq_destination_drop(char*receiver_name)
1688+
dmq_destination_drop(constchar*receiver_name)
16891689
{
16901690
DmqDestinationIddest_id;
16911691
pid_tsender_pid;

‎contrib/pg_exchange/dmq.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ extern void dmq_init(const char *library_name);
2424
externDmqDestinationIddmq_destination_add(char*connstr,char*sender_name,
2525
char*receiver_name,intping_period);
2626
externDmqConnStatedmq_get_destination_status(DmqDestinationIddest_id);
27-
externvoiddmq_destination_drop(char*receiver_name);
27+
externvoiddmq_destination_drop(constchar*receiver_name);
2828

2929
externDmqSenderIddmq_attach_receiver(constchar*sender_name,intmask_pos);
3030
externvoiddmq_detach_receiver(constchar*sender_name);
@@ -43,7 +43,7 @@ extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
4343
externvoiddmq_push(DmqDestinationIddest_id,char*stream_name,char*msg);
4444
externvoiddmq_push_buffer(DmqDestinationIddest_id,char*stream_name,constvoid*buffer,size_tlen);
4545

46-
typedefvoid (*dmq_receiver_hook_type) (char*);
46+
typedefvoid (*dmq_receiver_hook_type) (constchar*);
4747
externdmq_receiver_hook_typedmq_receiver_start_hook;
4848
externdmq_receiver_hook_typedmq_receiver_stop_hook;
4949

‎contrib/pg_exchange/pg_exchange.c

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
#include"nodeDummyscan.h"
2424
#include"nodes/nodes.h"
2525
#include"utils/builtins.h"
26+
#include"utils/hsearch.h"
2627
#include"utils/guc.h"
2728
#include"utils/memutils.h"
2829
#include"utils/plancache.h"
@@ -37,10 +38,12 @@ PG_FUNCTION_INFO_V1(pg_exec_plan);
3738

3839
#defineDMQ_CONNSTR_MAX_LEN 150
3940

41+
dmq_receiver_hook_typeold_dmq_receiver_stop_hook;
4042

4143
void_PG_init(void);
4244
staticvoiddeserialize_plan(char**squery,char**splan,char**sparams);
4345
staticvoidexec_plan(char*squery,PlannedStmt*pstmt,ParamListInfoparamLI,constchar*serverName);
46+
staticvoidOnNodeDisconnect(constchar*node_name);
4447

4548
staticSize
4649
shmem_size(void)
@@ -66,6 +69,9 @@ _PG_init(void)
6669

6770
RequestAddinShmemSpace(shmem_size());
6871
RequestNamedLWLockTranche("pg_exchange",1);
72+
73+
old_dmq_receiver_stop_hook=dmq_receiver_stop_hook;
74+
dmq_receiver_stop_hook=OnNodeDisconnect;
6975
}
7076

7177
Datum
@@ -198,3 +204,37 @@ exec_plan(char *squery, PlannedStmt *pstmt, ParamListInfo paramLI, const char *s
198204
receiver->rDestroy(receiver);
199205
ReleaseCachedPlan(cplan, false);
200206
}
207+
208+
staticvoid
209+
OnNodeDisconnect(constchar*node_name)
210+
{
211+
HASH_SEQ_STATUSstatus;
212+
DMQDestinations*dest;
213+
Oidserverid=InvalidOid;
214+
215+
elog(LOG,"Node %s: disconnected",node_name);
216+
217+
218+
LWLockAcquire(ExchShmem->lock,LW_EXCLUSIVE);
219+
220+
hash_seq_init(&status,ExchShmem->htab);
221+
222+
while ((dest=hash_seq_search(&status))!=NULL)
223+
{
224+
if (!(strcmp(dest->node,node_name)==0))
225+
continue;
226+
227+
serverid=dest->serverid;
228+
dmq_detach_receiver(node_name);
229+
dmq_destination_drop(node_name);
230+
break;
231+
}
232+
hash_seq_term(&status);
233+
234+
if (OidIsValid(serverid))
235+
hash_search(ExchShmem->htab,&serverid,HASH_REMOVE,NULL);
236+
else
237+
elog(LOG,"Record on disconnected server %u with name %s not found.",
238+
serverid,node_name);
239+
LWLockRelease(ExchShmem->lock);
240+
}

‎contrib/pg_exchange/stream.c

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -227,14 +227,12 @@ ISendTuple(DmqDestinationId dest_id, char *stream, TupleTableSlot *slot,
227227
staticvoid
228228
wait_for_delivery(OStream*ostream)
229229
{
230-
intattempts;
231-
232-
for (attempts=0;attempts<50;attempts++)
230+
for (;;)
233231
{
234232
intwaits;
235233

236234
pg_usleep(10);
237-
attempts++;
235+
238236
for (waits=0;waits<100000;waits++)
239237
{
240238
if (checkDelivery(ostream))

‎contrib/pg_execplan/tests/test.sql

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
SELECT*FROM pt;
2+
SELECT*FROM rt;
3+
SELECT*FROM st;
4+
SELECTcount(*)FROM pt;
5+
SELECTcount(*)FROM pt,rt;
6+
SELECTcount(*)FROM pt,rt,st;
7+
SELECTcount(*)FROM pt,rtWHEREpt.id=rt.id;
8+
SELECTcount(*)FROM pt,rt,stWHEREpt.id=rt.idandrt.id=st.id;
9+
SELECTcount(*)FROM pt,rt,stWHEREpt.id=rt.idandrt.id=st.payload;
10+
SELECTcount(*)FROM pt,rt,stWHEREpt.id=rt.payloadandrt.id=st.payload;
Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
export PGDATABASE=test_base
2+
psql -p 5432 -c"SELECT * FROM pt;"
3+
pg_ctl -D PGDATA_n1 stop
4+
pg_ctl -w -c -o"-p 5433" -D PGDATA_n1 -l n1.log start
5+
psql -p 5432 -c"SELECT * FROM pt;"

0 commit comments

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