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

Commitbc32577

Browse files
anna-akentevadlepikhova
authored andcommitted
PGPRO-4197: code cleanup + send stuff by 1024 bytes now
1 parent0b3be2b commitbc32577

File tree

3 files changed

+19
-42
lines changed

3 files changed

+19
-42
lines changed

‎pg_query_state.c

Lines changed: 12 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,9 @@ static void qs_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction,
6060
#endif
6161
staticvoidqs_ExecutorFinish(QueryDesc*queryDesc);
6262

63+
staticshm_mq_resultreceive_msg_by_parts(shm_mq_handle*mqh,Size*total,
64+
void**datap,boolnowait);
65+
6366
/* Global variables */
6467
List*QueryDescStack=NIL;
6568
staticProcSignalReasonUserIdPollReason=INVALID_PROCSIGNAL;
@@ -777,7 +780,7 @@ shm_mq_receive_with_timeout(shm_mq_handle *mqh,
777780
{
778781
shm_mq_resultmq_receive_result;
779782

780-
mq_receive_result=shm_mq_receive(mqh,nbytesp,datap, true);
783+
mq_receive_result=receive_msg_by_parts(mqh,nbytesp,datap, true);
781784
if (mq_receive_result!=SHM_MQ_WOULD_BLOCK)
782785
returnmq_receive_result;
783786
if (rc&WL_TIMEOUT||delay <=0)
@@ -960,51 +963,36 @@ copy_msg(shm_mq_msg *msg)
960963
returnresult;
961964
}
962965

963-
// ----------------- DEBUG -----------------
964-
staticvoid
965-
print_recv_bytes(intnum,char*src,intoffset)
966-
{
967-
elog(INFO,"======= RECV MSG SEGMENT START (%d bytes) =======",num);
968-
for (inti=offset;i<offset+num;i++)
969-
elog(INFO,"RECV byte #%d = %02x",i, (unsignedchar)*(src+i));
970-
}
971-
// ----------------- DEBUG -----------------
972-
973966
staticshm_mq_result
974-
shm_mq_receive_by_bytes(shm_mq_handle*mqh,Size*total,void**datap)
967+
receive_msg_by_parts(shm_mq_handle*mqh,Size*total,void**datap,
968+
boolnowait)
975969
{
976970
shm_mq_resultmq_receive_result;
977971
shm_mq_msg*buff;
978-
intii;
979972
intoffset;
980973
int*expected;
981974
Sizelen;
982975

983976
/* Get the expected number of bytes in message */
984-
mq_receive_result=shm_mq_receive(mqh,&len, (void**)&expected,false);
977+
mq_receive_result=shm_mq_receive(mqh,&len, (void**)&expected,nowait);
985978
if (mq_receive_result!=SHM_MQ_SUCCESS)
986979
returnmq_receive_result;
987980
Assert(len==sizeof(int));
988-
//elog(INFO, "======= RECV MSG (expecting %d bytes) =======", *expected);
989981

990982
*datap=palloc0(*expected);
991983

992984
/* Get the message itself */
993-
for (offset=0,ii=0;offset<*expected;ii++)
985+
for (offset=0;offset<*expected; )
994986
{
995-
// Keep receiving new messages until we assemble the full message
996-
mq_receive_result=shm_mq_receive(mqh,&len, ((void**)&buff),false);
987+
/* Keep receiving new messages until we assemble the full message */
988+
mq_receive_result=shm_mq_receive(mqh,&len, ((void**)&buff),nowait);
997989
memcpy((char*)*datap+offset,buff,len);
998-
//print_recv_bytes(len, (char *) *datap, offset);
999990
offset+=len;
1000991
if (mq_receive_result!=SHM_MQ_SUCCESS)
1001992
returnmq_receive_result;
1002993
}
1003994

1004-
//elog(INFO, "RECV: END cycle - %d", ii);
1005995
*total=offset;
1006-
//mq_receive_result = shm_mq_receive(mqh, &len, (void **) &msg, false);
1007-
//*datap = buff;
1008996

1009997
returnmq_receive_result;
1010998
}
@@ -1081,7 +1069,8 @@ GetRemoteBackendQueryStates(PGPROC *leader,
10811069
/* extract query state from leader process */
10821070
mqh=shm_mq_attach(mq,NULL,NULL);
10831071
elog(DEBUG1,"Wait response from leader %d",leader->pid);
1084-
mq_receive_result=shm_mq_receive_by_bytes(mqh,&len, ((void**)&msg));
1072+
mq_receive_result=receive_msg_by_parts(mqh,&len, (void**)&msg,
1073+
false);
10851074
if (mq_receive_result!=SHM_MQ_SUCCESS)
10861075
gotomq_error;
10871076
if (msg->reqid!=reqid)

‎pg_query_state.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
#include"storage/shm_mq.h"
1919

2020
#defineQUEUE_SIZE(16 * 1024)
21-
#defineBUF_SIZE7
21+
#defineMSG_MAX_SIZE1024
2222

2323
#defineTIMINIG_OFF_WARNING 1
2424
#defineBUFFERS_OFF_WARNING 2

‎signal_handler.c

Lines changed: 6 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ typedef struct
2727
char*plan;
2828
}stack_frame;
2929

30-
staticvoidsend_msg_by_bits(shm_mq_handle*mqh,Sizenbytes,constvoid*data);
30+
staticvoidsend_msg_by_parts(shm_mq_handle*mqh,Sizenbytes,constvoid*data);
3131

3232
/*
3333
*Get List of stack_frames as a stack of function calls starting from outermost call.
@@ -150,33 +150,21 @@ serialize_stack(char *dest, List *qs_stack)
150150
serialize_stack_frame(&dest,qs_frame);
151151
}
152152
}
153-
// ----------------- DEBUG -----------------
154-
staticvoid
155-
print_sent_bytes(intnum,char*src,intoffset)
156-
{
157-
elog(INFO,"======= SEND MSG SEGMENT START (%d bytes) =======",num);
158-
for (inti=offset;i<offset+num;i++)
159-
elog(INFO,"SENT byte #%d = %02x",i, (unsignedchar)*(src+i));
160-
}
161-
// ----------------- DEBUG -----------------
162153

163154
staticvoid
164-
send_msg_by_bits(shm_mq_handle*mqh,Sizenbytes,constvoid*data)
155+
send_msg_by_parts(shm_mq_handle*mqh,Sizenbytes,constvoid*data)
165156
{
166157
intbytes_left;
167158
intbytes_send;
168159

169160
/* Send the expected message length */
170161
shm_mq_send(mqh,sizeof(int),&nbytes, false);
171162

172-
//elog(INFO, "======= SEND MSG (%lu bytes) =======", nbytes);
173163
for (intoffset=0;offset<nbytes;offset+=bytes_send)
174164
{
175165
bytes_left=nbytes-offset;
176-
bytes_send= (bytes_left<BUF_SIZE) ?bytes_left :BUF_SIZE;
166+
bytes_send= (bytes_left<MSG_MAX_SIZE) ?bytes_left :MSG_MAX_SIZE;
177167
shm_mq_send(mqh,bytes_send,&(((unsignedchar*)data)[offset]), false);
178-
// DEBUG: print message that we just sent
179-
//print_sent_bytes(bytes_send, (char *) data, offset);
180168
}
181169
}
182170

@@ -238,15 +226,15 @@ SendQueryState(void)
238226
{
239227
shm_mq_msgmsg= {reqid,BASE_SIZEOF_SHM_MQ_MSG,MyProc,STAT_DISABLED };
240228

241-
send_msg_by_bits(mqh,msg.length,&msg);
229+
send_msg_by_parts(mqh,msg.length,&msg);
242230
}
243231

244232
/* check if backend doesn't execute any query */
245233
elseif (list_length(QueryDescStack)==0)
246234
{
247235
shm_mq_msgmsg= {reqid,BASE_SIZEOF_SHM_MQ_MSG,MyProc,QUERY_NOT_RUNNING };
248236

249-
send_msg_by_bits(mqh,msg.length,&msg);
237+
send_msg_by_parts(mqh,msg.length,&msg);
250238
}
251239

252240
/* happy path */
@@ -269,7 +257,7 @@ SendQueryState(void)
269257

270258
msg->stack_depth=list_length(qs_stack);
271259
serialize_stack(msg->stack,qs_stack);
272-
send_msg_by_bits(mqh,msglen,msg);
260+
send_msg_by_parts(mqh,msglen,msg);
273261
}
274262
elog(DEBUG1,"Worker %d sends response for pg_query_state to %d",shm_mq_get_sender(mq)->pid,shm_mq_get_receiver(mq)->pid);
275263
DetachPeer();

0 commit comments

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