multiple databases issue fixed

Author: zilder

dsm_array.c

@@ -258,6 +258,32 @@ free_dsm_array(DsmArray *arr)
 	arr->length = 0;
 }
 
+void
+resize_dsm_array(DsmArray *arr, size_t entry_size, size_t length)
+{
+	void *array_data;
+	size_t array_data_size;
+	void *buffer;
+
+	/* Copy data from array to temporary buffer */
+	array_data = dsm_array_get_pointer(arr);
+	array_data_size = arr->length * entry_size;
+	buffer = palloc(array_data_size);
+	memcpy(buffer, array_data, array_data_size);
+
+	/* Free array */
+	free_dsm_array(arr);
+
+	/* Allocate new array */
+	alloc_dsm_array(arr, entry_size, length);
+
+	/* Copy data to new array */
+	array_data = dsm_array_get_pointer(arr);
+	memcpy(array_data, buffer, array_data_size);
+
+	pfree(buffer);
+}
+
 void *
 dsm_array_get_pointer(const DsmArray* arr)
 {

init.c

@@ -75,17 +75,58 @@ void
 load_config(void)
 {
 	bool new_segment_created;
+	Oid *databases;
 
 	initialization_needed = false;
 	new_segment_created = init_dsm_segment(INITIAL_BLOCKS_COUNT, 32);
 
-	/* if config is not loaded */
+	/* If dsm segment just created */
 	if (new_segment_created)
 	{
-		LWLockAcquire(pmstate->load_config_lock, LW_EXCLUSIVE);
-		load_relations_hashtable(new_segment_created);
-		LWLockRelease(pmstate->load_config_lock);
+		/*
+		 * Allocate databases array and put current database
+		 * oid into it. This array contains databases oids
+		 * that have already been cached (to prevent repeat caching)
+		 */
+		LWLockAcquire(pmstate->dsm_init_lock, LW_EXCLUSIVE);
+
+		if (&pmstate->databases.length > 0)
+			free_dsm_array(&pmstate->databases);
+		alloc_dsm_array(&pmstate->databases, sizeof(Oid), 1);
+		databases = (Oid *) dsm_array_get_pointer(&pmstate->databases);
+		databases[0] = MyDatabaseId;
+
+		LWLockRelease(pmstate->dsm_init_lock);
+	}
+	else
+	{
+		int databases_count = pmstate->databases.length;
+		int i;
+
+		/* Check if we already cached config for current database */
+		LWLockAcquire(pmstate->dsm_init_lock, LW_EXCLUSIVE);
+
+		databases = (Oid *) dsm_array_get_pointer(&pmstate->databases);
+		for(i=0; i<databases_count; i++)
+			if (databases[i] == MyDatabaseId)
+			{
+				// LWLockRelease(pmstate->load_config_lock);
+				LWLockRelease(pmstate->dsm_init_lock);
+				return;
+			}
+
+		/* Put current database oid to databases list */
+		resize_dsm_array(&pmstate->databases, sizeof(Oid), databases_count + 1);
+		databases = (Oid *) dsm_array_get_pointer(&pmstate->databases);
+		databases[databases_count] = MyDatabaseId;
+
+		LWLockRelease(pmstate->dsm_init_lock);
 	}
+
+	/* Load cache */
+	LWLockAcquire(pmstate->load_config_lock, LW_EXCLUSIVE);
+	load_relations_hashtable(new_segment_created);
+	LWLockRelease(pmstate->load_config_lock);
 }
 
 /*

pathman.h

@@ -109,9 +109,10 @@ typedef struct RangeRelation
 
 typedef struct PathmanState
 {
-	LWLock   *load_config_lock;
-	LWLock   *dsm_init_lock;
-	LWLock   *edit_partitions_lock;
+	LWLock	   *load_config_lock;
+	LWLock	   *dsm_init_lock;
+	LWLock	   *edit_partitions_lock;
+	DsmArray	databases;
 } PathmanState;
 
 PathmanState *pmstate;
@@ -156,6 +157,7 @@ bool init_dsm_segment(size_t blocks_count, size_t block_size);
 void init_dsm_table(size_t block_size, size_t start, size_t end);
 void alloc_dsm_array(DsmArray *arr, size_t entry_size, size_t length);
 void free_dsm_array(DsmArray *arr);
+void resize_dsm_array(DsmArray *arr, size_t entry_size, size_t length);
 void *dsm_array_get_pointer(const DsmArray* arr);
 dsm_handle get_dsm_array_segment(void);
 void attach_dsm_array_segment(void);