Remove unnecessary overhead in backend's large-object operations.

Do read/write permissions checks at most once per large object descriptor,
not once per lo_read or lo_write call as before.  The repeated tests were
quite useless in the read case since the snapshot-based tests were
guaranteed to produce the same answer every time.  In the write case,
the extra tests could in principle detect revocation of write privileges
after a series of writes has started --- but there's a race condition there
anyway, since we'd check privileges before performing and certainly before
committing the write.  So there's no real advantage to checking every
single time, and we might as well redefine it as "only check the first
time".

On the same reasoning, remove the LargeObjectExists checks in inv_write
and inv_truncate.  We already checked existence when the descriptor was
opened, and checking again doesn't provide any real increment of safety
that would justify the cost.

diff --git a/src/backend/libpq/be-fsstubs.c b/src/backend/libpq/be-fsstubs.c
index c4ac1a650f505cfa1582a19797cf9621a4c80f46..dbc00b450d1890102ae00683bca5b5fab1976e8d 100644 (file)
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -157,24 +157,32 @@ int
 lo_read(int fd, char *buf, int len)
 {
        int                     status;
+       LargeObjectDesc *lobj;
 
        if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
                ereport(ERROR,
                                (errcode(ERRCODE_UNDEFINED_OBJECT),
                                 errmsg("invalid large-object descriptor: %d", fd)));
+       lobj = cookies[fd];
 
-       /* Permission checks */
-       if (!lo_compat_privileges &&
-               pg_largeobject_aclcheck_snapshot(cookies[fd]->id,
-                                                                                GetUserId(),
-                                                                                ACL_SELECT,
-                                                                          cookies[fd]->snapshot) != ACLCHECK_OK)
-               ereport(ERROR,
-                               (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-                                errmsg("permission denied for large object %u",
-                                               cookies[fd]->id)));
+       /* We don't bother to check IFS_RDLOCK, since it's always set */
+
+       /* Permission checks --- first time through only */
+       if ((lobj->flags & IFS_RD_PERM_OK) == 0)
+       {
+               if (!lo_compat_privileges &&
+                       pg_largeobject_aclcheck_snapshot(lobj->id,
+                                                                                        GetUserId(),
+                                                                                        ACL_SELECT,
+                                                                                        lobj->snapshot) != ACLCHECK_OK)
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+                                        errmsg("permission denied for large object %u",
+                                                       lobj->id)));
+               lobj->flags |= IFS_RD_PERM_OK;
+       }
 
-       status = inv_read(cookies[fd], buf, len);
+       status = inv_read(lobj, buf, len);
 
        return status;
 }
@@ -183,30 +191,36 @@ int
 lo_write(int fd, const char *buf, int len)
 {
        int                     status;
+       LargeObjectDesc *lobj;
 
        if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
                ereport(ERROR,
                                (errcode(ERRCODE_UNDEFINED_OBJECT),
                                 errmsg("invalid large-object descriptor: %d", fd)));
+       lobj = cookies[fd];
 
-       if ((cookies[fd]->flags & IFS_WRLOCK) == 0)
+       if ((lobj->flags & IFS_WRLOCK) == 0)
                ereport(ERROR,
                                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                          errmsg("large object descriptor %d was not opened for writing",
                                         fd)));
 
-       /* Permission checks */
-       if (!lo_compat_privileges &&
-               pg_largeobject_aclcheck_snapshot(cookies[fd]->id,
-                                                                                GetUserId(),
-                                                                                ACL_UPDATE,
-                                                                          cookies[fd]->snapshot) != ACLCHECK_OK)
-               ereport(ERROR,
-                               (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-                                errmsg("permission denied for large object %u",
-                                               cookies[fd]->id)));
+       /* Permission checks --- first time through only */
+       if ((lobj->flags & IFS_WR_PERM_OK) == 0)
+       {
+               if (!lo_compat_privileges &&
+                       pg_largeobject_aclcheck_snapshot(lobj->id,
+                                                                                        GetUserId(),
+                                                                                        ACL_UPDATE,
+                                                                                        lobj->snapshot) != ACLCHECK_OK)
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+                                        errmsg("permission denied for large object %u",
+                                                       lobj->id)));
+               lobj->flags |= IFS_WR_PERM_OK;
+       }
 
-       status = inv_write(cookies[fd], buf, len);
+       status = inv_write(lobj, buf, len);
 
        return status;
 }
@@ -230,8 +244,8 @@ lo_lseek(PG_FUNCTION_ARGS)
        if (status != (int32) status)
                ereport(ERROR,
                                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
-                                errmsg("lo_lseek result out of range for large-object descriptor %d",
-                                               fd)));
+               errmsg("lo_lseek result out of range for large-object descriptor %d",
+                          fd)));
 
        PG_RETURN_INT32((int32) status);
 }
@@ -303,8 +317,8 @@ lo_tell(PG_FUNCTION_ARGS)
        if (offset != (int32) offset)
                ereport(ERROR,
                                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
-                                errmsg("lo_tell result out of range for large-object descriptor %d",
-                                               fd)));
+                errmsg("lo_tell result out of range for large-object descriptor %d",
+                               fd)));
 
        PG_RETURN_INT32((int32) offset);
 }
@@ -558,30 +572,48 @@ lo_export(PG_FUNCTION_ARGS)
  * lo_truncate -
  *       truncate a large object to a specified length
  */
-Datum
-lo_truncate(PG_FUNCTION_ARGS)
+static void
+lo_truncate_internal(int32 fd, int64 len)
 {
-       int32           fd = PG_GETARG_INT32(0);
-       int32           len = PG_GETARG_INT32(1);
+       LargeObjectDesc *lobj;
 
        if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
                ereport(ERROR,
                                (errcode(ERRCODE_UNDEFINED_OBJECT),
                                 errmsg("invalid large-object descriptor: %d", fd)));
+       lobj = cookies[fd];
 
-       /* Permission checks */
-       if (!lo_compat_privileges &&
-               pg_largeobject_aclcheck_snapshot(cookies[fd]->id,
-                                                                                GetUserId(),
-                                                                                ACL_UPDATE,
-                                                                          cookies[fd]->snapshot) != ACLCHECK_OK)
+       if ((lobj->flags & IFS_WRLOCK) == 0)
                ereport(ERROR,
-                               (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-                                errmsg("permission denied for large object %u",
-                                               cookies[fd]->id)));
+                               (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                         errmsg("large object descriptor %d was not opened for writing",
+                                        fd)));
+
+       /* Permission checks --- first time through only */
+       if ((lobj->flags & IFS_WR_PERM_OK) == 0)
+       {
+               if (!lo_compat_privileges &&
+                       pg_largeobject_aclcheck_snapshot(lobj->id,
+                                                                                        GetUserId(),
+                                                                                        ACL_UPDATE,
+                                                                                        lobj->snapshot) != ACLCHECK_OK)
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+                                        errmsg("permission denied for large object %u",
+                                                       lobj->id)));
+               lobj->flags |= IFS_WR_PERM_OK;
+       }
 
-       inv_truncate(cookies[fd], len);
+       inv_truncate(lobj, len);
+}
 
+Datum
+lo_truncate(PG_FUNCTION_ARGS)
+{
+       int32           fd = PG_GETARG_INT32(0);
+       int32           len = PG_GETARG_INT32(1);
+
+       lo_truncate_internal(fd, len);
        PG_RETURN_INT32(0);
 }
 
@@ -591,24 +623,7 @@ lo_truncate64(PG_FUNCTION_ARGS)
        int32           fd = PG_GETARG_INT32(0);
        int64           len = PG_GETARG_INT64(1);
 
-       if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_OBJECT),
-                                errmsg("invalid large-object descriptor: %d", fd)));
-
-       /* Permission checks */
-       if (!lo_compat_privileges &&
-               pg_largeobject_aclcheck_snapshot(cookies[fd]->id,
-                                                                                GetUserId(),
-                                                                                ACL_UPDATE,
-                                                                          cookies[fd]->snapshot) != ACLCHECK_OK)
-               ereport(ERROR,
-                               (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-                                errmsg("permission denied for large object %u",
-                                               cookies[fd]->id)));
-
-       inv_truncate(cookies[fd], len);
-
+       lo_truncate_internal(fd, len);
        PG_RETURN_INT32(0);
 }
 

diff --git a/src/backend/storage/large_object/inv_api.c b/src/backend/storage/large_object/inv_api.c
index 5dd17823c2312f088a4a6c64ad58ab7668f2e1b2..ad8424b9bfcb3ab020da04c6b5bf2dd94aaec58a 100644 (file)
--- a/src/backend/storage/large_object/inv_api.c
+++ b/src/backend/storage/large_object/inv_api.c
@@ -420,8 +420,8 @@ inv_seek(LargeObjectDesc *obj_desc, int64 offset, int whence)
        if (newoffset < 0 || newoffset > MAX_LARGE_OBJECT_SIZE)
                ereport(ERROR,
                                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-                                errmsg_internal("invalid large object seek target: " INT64_FORMAT,
-                                                                newoffset)));
+                  errmsg_internal("invalid large object seek target: " INT64_FORMAT,
+                                                  newoffset)));
 
        obj_desc->offset = newoffset;
        return newoffset;
@@ -562,18 +562,7 @@ inv_write(LargeObjectDesc *obj_desc, const char *buf, int nbytes)
        Assert(buf != NULL);
 
        /* enforce writability because snapshot is probably wrong otherwise */
-       if ((obj_desc->flags & IFS_WRLOCK) == 0)
-               ereport(ERROR,
-                               (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-                                errmsg("large object %u was not opened for writing",
-                                               obj_desc->id)));
-
-       /* check existence of the target largeobject */
-       if (!LargeObjectExists(obj_desc->id))
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_OBJECT),
-                                errmsg("large object %u was already dropped",
-                                               obj_desc->id)));
+       Assert(obj_desc->flags & IFS_WRLOCK);
 
        if (nbytes <= 0)
                return 0;
@@ -767,18 +756,7 @@ inv_truncate(LargeObjectDesc *obj_desc, int64 len)
        Assert(PointerIsValid(obj_desc));
 
        /* enforce writability because snapshot is probably wrong otherwise */
-       if ((obj_desc->flags & IFS_WRLOCK) == 0)
-               ereport(ERROR,
-                               (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-                                errmsg("large object %u was not opened for writing",
-                                               obj_desc->id)));
-
-       /* check existence of the target largeobject */
-       if (!LargeObjectExists(obj_desc->id))
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_OBJECT),
-                                errmsg("large object %u was already dropped",
-                                               obj_desc->id)));
+       Assert(obj_desc->flags & IFS_WRLOCK);
 
        /*
         * use errmsg_internal here because we don't want to expose INT64_FORMAT

diff --git a/src/include/storage/large_object.h b/src/include/storage/large_object.h
index f8232411a753e4d057409636af6b9a2c05d8b484..77a7a3dde8b559d7328c013c3532aee8f356cea7 100644 (file)
--- a/src/include/storage/large_object.h
+++ b/src/include/storage/large_object.h
@@ -27,6 +27,10 @@
  * offset is the current seek offset within the LO
  * flags contains some flag bits
  *
+ * NOTE: in current usage, flag bit IFS_RDLOCK is *always* set, and we don't
+ * bother to test for it.  Permission checks are made at first read or write
+ * attempt, not during inv_open(), so we have other bits to remember that.
+ *
  * NOTE: before 7.1, we also had to store references to the separate table
  * and index of a specific large object.  Now they all live in pg_largeobject
  * and are accessed via a common relation descriptor.
@@ -38,11 +42,13 @@ typedef struct LargeObjectDesc
        Snapshot        snapshot;               /* snapshot to use */
        SubTransactionId subid;         /* owning subtransaction ID */
        uint64          offset;                 /* current seek pointer */
-       int                     flags;                  /* locking info, etc */
+       int                     flags;                  /* see flag bits below */
 
-/* flag bits: */
-#define IFS_RDLOCK             (1 << 0)
-#define IFS_WRLOCK             (1 << 1)
+/* bits in flags: */
+#define IFS_RDLOCK             (1 << 0)        /* LO was opened for reading */
+#define IFS_WRLOCK             (1 << 1)        /* LO was opened for writing */
+#define IFS_RD_PERM_OK (1 << 2)        /* read permission has been verified */
+#define IFS_WR_PERM_OK (1 << 3)        /* write permission has been verified */
 
 } LargeObjectDesc;