diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 3f5f054..6cb2d44 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -351,8 +351,10 @@ static List *upperPendingActions = NIL; /* list of upper-xact lists */ */ typedef struct Notification { - char *channel; /* channel name */ - char *payload; /* payload string (can be empty) */ + uint16 channel_len; /* length of channel-name string */ + uint16 payload_len; /* length of payload string */ + /* null-terminated channel name, then null-terminated payload follow */ + char data[FLEXIBLE_ARRAY_MEMBER]; } Notification; typedef struct NotificationList @@ -417,7 +419,7 @@ static bool asyncQueueProcessPageEntries(volatile QueuePosition *current, Snapshot snapshot); static void asyncQueueAdvanceTail(void); static void ProcessIncomingNotify(void); -static bool AsyncExistsPendingNotify(const char *channel, const char *payload); +static bool AsyncExistsPendingNotify(Notification *n); static void AddEventToPendingNotifies(Notification *n); static uint32 notification_hash(const void *key, Size keysize); static int notification_match(const void *key1, const void *key2, Size keysize); @@ -569,6 +571,8 @@ pg_notify(PG_FUNCTION_ARGS) void Async_Notify(const char *channel, const char *payload) { + size_t channel_len; + size_t payload_len; Notification *n; MemoryContext oldcontext; @@ -578,41 +582,53 @@ Async_Notify(const char *channel, const char *payload) if (Trace_notify) elog(DEBUG1, "Async_Notify(%s)", channel); + channel_len = channel ? strlen(channel) : 0; + payload_len = payload ? strlen(payload) : 0; + /* a channel name must be specified */ - if (!channel || !strlen(channel)) + if (channel_len == 0) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("channel name cannot be empty"))); - if (strlen(channel) >= NAMEDATALEN) + /* enforce length limits */ + if (channel_len >= NAMEDATALEN) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("channel name too long"))); - if (payload) - { - if (strlen(payload) >= NOTIFY_PAYLOAD_MAX_LENGTH) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("payload string too long"))); - } - - /* no point in making duplicate entries in the list ... */ - if (AsyncExistsPendingNotify(channel, payload)) - return; + if (payload_len >= NOTIFY_PAYLOAD_MAX_LENGTH) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("payload string too long"))); /* + * We must construct the Notification entry, even if we end up not using + * it, in order to compare it cheaply to existing list entries. + * * The notification list needs to live until end of transaction, so store * it in the transaction context. */ oldcontext = MemoryContextSwitchTo(CurTransactionContext); - n = (Notification *) palloc(sizeof(Notification)); - n->channel = pstrdup(channel); + n = (Notification *) palloc(offsetof(Notification, data) + + channel_len + payload_len + 2); + n->channel_len = channel_len; + n->payload_len = payload_len; + strcpy(n->data, channel); if (payload) - n->payload = pstrdup(payload); + strcpy(n->data + channel_len + 1, payload); else - n->payload = ""; + n->data[channel_len + 1] = '\0'; + + /* Now check for duplicates */ + if (AsyncExistsPendingNotify(n)) + { + /* It's a dup, so forget it */ + pfree(n); + MemoryContextSwitchTo(oldcontext); + return; + } if (pendingNotifies == NULL) { @@ -1303,8 +1319,8 @@ asyncQueueAdvance(volatile QueuePosition *position, int entryLength) static void asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe) { - size_t channellen = strlen(n->channel); - size_t payloadlen = strlen(n->payload); + size_t channellen = n->channel_len; + size_t payloadlen = n->payload_len; int entryLength; Assert(channellen < NAMEDATALEN); @@ -1317,8 +1333,7 @@ asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe) qe->dboid = MyDatabaseId; qe->xid = GetCurrentTransactionId(); qe->srcPid = MyProcPid; - memcpy(qe->data, n->channel, channellen + 1); - memcpy(qe->data + channellen + 1, n->payload, payloadlen + 1); + memcpy(qe->data, n->data, channellen + payloadlen + 2); } /* @@ -1701,7 +1716,7 @@ AtSubCommit_Notify(void) { Notification *childn = (Notification *) lfirst(l); - if (!AsyncExistsPendingNotify(childn->channel, childn->payload)) + if (!AsyncExistsPendingNotify(childn)) AddEventToPendingNotifies(childn); } } @@ -2159,29 +2174,18 @@ NotifyMyFrontEnd(const char *channel, const char *payload, int32 srcPid) elog(INFO, "NOTIFY for \"%s\" payload \"%s\"", channel, payload); } -/* Does pendingNotifies include the given channel/payload? */ +/* Does pendingNotifies include a match for the given event? */ static bool -AsyncExistsPendingNotify(const char *channel, const char *payload) +AsyncExistsPendingNotify(Notification *n) { if (pendingNotifies == NULL) return false; - if (payload == NULL) - payload = ""; - if (pendingNotifies->hashtab != NULL) { /* Use the hash table to probe for a match */ - Notification n; - Notification *k; - - /* set up a dummy Notification struct */ - n.channel = unconstify(char *, channel); - n.payload = unconstify(char *, payload); - k = &n; - /* ... and probe */ if (hash_search(pendingNotifies->hashtab, - &k, + &n, HASH_FIND, NULL)) return true; @@ -2193,10 +2197,12 @@ AsyncExistsPendingNotify(const char *channel, const char *payload) foreach(l, pendingNotifies->events) { - Notification *n = (Notification *) lfirst(l); + Notification *oldn = (Notification *) lfirst(l); - if (strcmp(n->channel, channel) == 0 && - strcmp(n->payload, payload) == 0) + if (n->channel_len == oldn->channel_len && + n->payload_len == oldn->payload_len && + memcmp(n->data, oldn->data, + n->channel_len + n->payload_len + 2) == 0) return true; } } @@ -2278,16 +2284,11 @@ static uint32 notification_hash(const void *key, Size keysize) { const Notification *k = *(const Notification *const *) key; - uint32 hashc; - uint32 hashp; Assert(keysize == sizeof(Notification *)); - /* We just XOR the hashes for the two strings */ - hashc = DatumGetUInt32(hash_any((const unsigned char *) k->channel, - (int) strlen((const char *) k->channel))); - hashp = DatumGetUInt32(hash_any((const unsigned char *) k->payload, - (int) strlen((const char *) k->payload))); - return hashc ^ hashp; + /* We don't bother to include the payload's trailing null in the hash */ + return DatumGetUInt32(hash_any((const unsigned char *) k->data, + k->channel_len + k->payload_len + 1)); } /* @@ -2300,8 +2301,10 @@ notification_match(const void *key1, const void *key2, Size keysize) const Notification *k2 = *(const Notification *const *) key2; Assert(keysize == sizeof(Notification *)); - if (strcmp(k1->channel, k2->channel) == 0 && - strcmp(k1->payload, k2->payload) == 0) + if (k1->channel_len == k2->channel_len && + k1->payload_len == k2->payload_len && + memcmp(k1->data, k2->data, + k1->channel_len + k1->payload_len + 2) == 0) return 0; /* equal */ return 1; /* not equal */ }