Index: trunk/kernel/mm/ppm.c
===================================================================
--- trunk/kernel/mm/ppm.c	(revision 625)
+++ trunk/kernel/mm/ppm.c	(revision 632)
@@ -3,5 +3,5 @@
  *
  * Authors  Ghassan Almaless (2008,2009,2010,2011,2012)
- *          Alain Greiner    (2016,2017,2018)
+ *          Alain Greiner    (2016,2017,2018,2019)
  *
  * Copyright (c) UPMC Sorbonne Universites
@@ -45,11 +45,4 @@
 ////////////////////////////////////////////////////////////////////////////////////////
 
-////////////////////////////////////////////////
-inline bool_t ppm_page_is_valid( page_t * page )
-{
-	ppm_t    * ppm  = &LOCAL_CLUSTER->ppm;
-	uint32_t   pgnr = (uint32_t)( page - ppm->pages_tbl );
-	return (pgnr <= ppm->pages_nr);
-}
 
 /////////////////////////////////////////////
@@ -151,18 +144,18 @@
 void ppm_free_pages_nolock( page_t * page )
 {
-	page_t   * buddy;            // searched buddy page descriptor
-	uint32_t   buddy_index;      // buddy page index
-	page_t   * current;          // current (merged) page descriptor
-	uint32_t   current_index;    // current (merged) page index
-	uint32_t   current_order;    // current (merged) page order
+	page_t   * buddy;            // searched buddy block page descriptor
+	uint32_t   buddy_index;      // buddy bloc index in page_tbl[]
+	page_t   * current;          // current (merged) block page descriptor
+	uint32_t   current_index;    // current (merged) block index in page_tbl[]
+	uint32_t   current_order;    // current (merged) block order
 
 	ppm_t    * ppm         = &LOCAL_CLUSTER->ppm;
 	page_t   * pages_tbl   = ppm->pages_tbl;
 
-	assert( !page_is_flag( page , PG_FREE ) ,
-    "page already released : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
-
-	assert( !page_is_flag( page , PG_RESERVED ) ,
-    "reserved page : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
+assert( !page_is_flag( page , PG_FREE ) ,
+"page already released : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
+
+assert( !page_is_flag( page , PG_RESERVED ) ,
+"reserved page : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
 
 	// update released page descriptor flags
@@ -172,5 +165,5 @@
 	// - merge with current page descriptor if found
 	// - exit to release the current page descriptor if not found
-	current       = page ,
+	current       = page;
 	current_index = (uint32_t)(page - ppm->pages_tbl);
 	for( current_order = page->order ;
@@ -181,20 +174,24 @@
 		buddy       = pages_tbl + buddy_index;
 
-		if( !page_is_flag( buddy , PG_FREE ) || (buddy->order != current_order) ) break;
-
-		// remove buddy from free list
+        // exit this loop if buddy block not found
+		if( !page_is_flag( buddy , PG_FREE ) || 
+            (buddy->order != current_order) ) break;
+
+		// remove buddy block from free_list 
 		list_unlink( &buddy->list );
 		ppm->free_pages_nr[current_order] --;
 
-		// merge buddy with current
+        // reset order field in buddy block page descriptor
 		buddy->order = 0;
+
+		// compute merged block index in page_tbl[]
 		current_index &= buddy_index;
 	}
 
-	// update merged page descriptor order
+	// update pointer and order field for merged block page descriptor 
 	current        = pages_tbl + current_index;
 	current->order = current_order;
 
-	// insert current in free list
+	// insert merged block in free list
 	list_add_first( &ppm->free_pages_root[current_order] , &current->list );
 	ppm->free_pages_nr[current_order] ++;
@@ -205,7 +202,8 @@
 page_t * ppm_alloc_pages( uint32_t   order )
 {
+	page_t   * current_block;
 	uint32_t   current_order;
-	page_t   * remaining_block;
 	uint32_t   current_size;
+	page_t   * found_block;  
 
 #if DEBUG_PPM_ALLOC_PAGES
@@ -213,11 +211,11 @@
 uint32_t cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_PPM_ALLOC_PAGES < cycle )
-printk("\n[%s] thread[%x,%x] enter for %d page(s) / cycle %d\n",
-__FUNCTION__, this->process->pid, this->trdid, 1<<order, cycle );
+printk("\n[%s] thread[%x,%x] enter for %d page(s) in cluster %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 1<<order, cxy, cycle );
 #endif
 
 #if(DEBUG_PPM_ALLOC_PAGES & 0x1)
 if( DEBUG_PPM_ALLOC_PAGES < cycle )
-ppm_print("enter ppm_alloc_pages");
+ppm_remote_display( local_cxy );
 #endif
 
@@ -227,8 +225,11 @@
 assert( (order < CONFIG_PPM_MAX_ORDER) , "illegal order argument = %d\n" , order );
 
-	page_t * block = NULL;  
+    //build extended pointer on lock protecting remote PPM
+    xptr_t lock_xp = XPTR( local_cxy , &ppm->free_lock );
 
 	// take lock protecting free lists
-	busylock_acquire( &ppm->free_lock );
+	remote_busylock_acquire( lock_xp );
+
+	current_block = NULL;
 
 	// find a free block equal or larger to requested size
@@ -237,20 +238,33 @@
 		if( !list_is_empty( &ppm->free_pages_root[current_order] ) )
 		{
-			block = LIST_FIRST( &ppm->free_pages_root[current_order] , page_t , list );
-			list_unlink( &block->list );
-			break;
+            // get first free block in this free_list
+			current_block = LIST_FIRST( &ppm->free_pages_root[current_order] , page_t , list );
+
+            // remove this block from this free_list
+			list_unlink( &current_block->list );
+
+            // register pointer on found block
+            found_block = current_block;
+
+            // update this free-list number of blocks
+	        ppm->free_pages_nr[current_order] --;
+
+            // compute found block size
+	        current_size = (1 << current_order);
+
+			break;	
 		}
 	}
 
-	if( block == NULL ) // return failure
+	if( current_block == NULL ) // return failure if no free block found
 	{
 		// release lock protecting free lists
-		busylock_release( &ppm->free_lock );
+		remote_busylock_release( lock_xp );
 
 #if DEBUG_PPM_ALLOC_PAGES
 cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_PPM_ALLOC_PAGES < cycle )
-printk("\n[%s] thread[%x,%x] cannot allocate %d page(s) / cycle %d\n",
-__FUNCTION__, this->process->pid, this->trdid, 1<<order, cycle );
+printk("\n[%s] thread[%x,%x] cannot allocate %d page(s) in cluster %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 1<<order, cxy, cycle );
 #endif
 
@@ -258,47 +272,48 @@
 	}
 
-	// update free-lists after removing a block
-	ppm->free_pages_nr[current_order] --;
-	current_size = (1 << current_order);
-
-	// split the removed block in smaller sub-blocks if required
+
+	// split the found block in smaller sub-blocks if required
 	// and update the free-lists accordingly
 	while( current_order > order )
 	{
 		current_order --;
+
+        // update pointer, size, and order fiels for new free block
 		current_size >>= 1;
-
-		remaining_block = block + current_size;
-		remaining_block->order = current_order;
-
-		list_add_first( &ppm->free_pages_root[current_order] , &remaining_block->list );
+		current_block = found_block + current_size;
+		current_block->order = current_order;
+
+        // insert new free block in relevant free_list
+		list_add_first( &ppm->free_pages_root[current_order] , &current_block->list );
+
+        // update number of blocks in free list
 		ppm->free_pages_nr[current_order] ++;
 	}
 
-	// update page descriptor
-	page_clear_flag( block , PG_FREE );
-	page_refcount_up( block );
-	block->order = order;
+	// update found block page descriptor
+	page_clear_flag( found_block , PG_FREE );
+	page_refcount_up( found_block );
+	found_block->order = order;
 
 	// release lock protecting free lists
-	busylock_release( &ppm->free_lock );
+	remote_busylock_release( lock_xp );
 
     // update DQDT
-    dqdt_increment_pages( order );
+    dqdt_increment_pages( local_cxy , order );
 
 #if DEBUG_PPM_ALLOC_PAGES
 cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_PPM_ALLOC_PAGES < cycle )
-printk("\n[%s] thread[%x,%x] exit for %d page(s) / ppn = %x / cycle %d\n",
+printk("\n[%s] thread[%x,%x] exit for %d page(s) in cluster %x / ppn = %x / cycle %d\n",
 __FUNCTION__, this->process->pid, this->trdid, 
-1<<order, ppm_page2ppn(XPTR( local_cxy , block )), cycle );
+1<<order, ppm_page2ppn(XPTR( local_cxy , found_block )), cxy, cycle );
 #endif
 
 #if(DEBUG_PPM_ALLOC_PAGES & 0x1)
 if( DEBUG_PPM_ALLOC_PAGES < cycle )
-ppm_print("exit ppm_alloc_pages");
-#endif
-
-	return block;
+ppm_remote_display( local_cxy );
+#endif
+
+	return found_block;
 
 }  // end ppm_alloc_pages()
@@ -311,44 +326,289 @@
 
 #if DEBUG_PPM_FREE_PAGES
-uint32_t cycle = (uint32_t)hal_get_cycles();
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_PPM_FREE_PAGES < cycle )
-printk("\n[%s] thread[%x,%x] enter for %d page(s) / ppn %x / cycle %d\n",
+printk("\n[%s] thread[%x,%x] enter for %d page(s) in cluster %x / ppn %x / cycle %d\n",
 __FUNCTION__, this->process->pid, this->trdid, 
-1<<page->order, ppm_page2ppn(XPTR(local_cxy , page)), cycle );
+1<<page->order, local_cxy, ppm_page2ppn(XPTR(local_cxy , page)), cycle );
 #endif
 
 #if(DEBUG_PPM_FREE_PAGES & 0x1)
 if( DEBUG_PPM_FREE_PAGES < cycle )
-ppm_print("enter ppm_free_pages");
-#endif
+ppm_remote_display( local_cxy );
+#endif
+
+    //build extended pointer on lock protecting free_lists
+    xptr_t lock_xp = XPTR( local_cxy , &ppm->free_lock );
 
 	// get lock protecting free_pages[] array
-	busylock_acquire( &ppm->free_lock );
+	remote_busylock_acquire( lock_xp );
 
 	ppm_free_pages_nolock( page );
 
-	// release lock protecting free_pages[] array
-	busylock_release( &ppm->free_lock );
+	// release lock protecting free_lists
+	remote_busylock_release( lock_xp );
 
     // update DQDT
-    dqdt_decrement_pages( page->order );
+    dqdt_decrement_pages( local_cxy , page->order );
 
 #if DEBUG_PPM_FREE_PAGES
 cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_PPM_FREE_PAGES < cycle )
-printk("\n[%s] thread[%x,%x] exit for %d page(s) / ppn %x / cycle %d\n",
+printk("\n[%s] thread[%x,%x] exit for %d page(s) in cluster %x / ppn %x / cycle %d\n",
 __FUNCTION__, this->process->pid, this->trdid, 
-1<<page->order, ppm_page2ppn(XPTR(local_cxy , page)), cycle );
+1<<page->order, local_cxy, ppm_page2ppn(XPTR(local_cxy , page)) , cycle );
 #endif
 
 #if(DEBUG_PPM_FREE_PAGES & 0x1)
 if( DEBUG_PPM_FREE_PAGES < cycle )
-ppm_print("exit ppm_free_pages");
+ppm_remote_display( local_cxy );
 #endif
 
 }  // end ppm_free_pages()
 
-////////////////////////
-void ppm_display( void )
+/////////////////////////////////////////////
+xptr_t ppm_remote_alloc_pages( cxy_t     cxy,
+                               uint32_t  order )
+{
+	uint32_t   current_order;
+	uint32_t   current_size;
+    page_t   * current_block;   
+    page_t   * found_block;
+
+#if DEBUG_PPM_ALLOC_PAGES
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_PPM_ALLOC_PAGES < cycle )
+printk("\n[%s] thread[%x,%x] enter for %d small page(s) in cluster %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 1<<order, cxy, cycle );
+#endif
+
+#if(DEBUG_PPM_ALLOC_PAGES & 0x1)
+if( DEBUG_PPM_ALLOC_PAGES < cycle )
+ppm_remote_display( cxy );
+#endif
+
+// check order
+assert( (order < CONFIG_PPM_MAX_ORDER) , "illegal order argument = %d\n" , order );
+
+    // get local pointer on PPM (same in all clusters)
+	ppm_t * ppm = &LOCAL_CLUSTER->ppm;
+
+    //build extended pointer on lock protecting remote PPM
+    xptr_t lock_xp = XPTR( cxy , &ppm->free_lock );
+
+	// take lock protecting free lists in remote cluster
+	remote_busylock_acquire( lock_xp );
+
+    current_block = NULL;    
+
+	// find in remote cluster a free block equal or larger to requested size
+	for( current_order = order ; current_order < CONFIG_PPM_MAX_ORDER ; current_order ++ )
+	{
+        // get local pointer on the root of relevant free_list in remote cluster
+        list_entry_t * root = &ppm->free_pages_root[current_order];
+
+		if( !list_remote_is_empty( cxy , root ) )
+		{
+            // get local pointer on first free page descriptor in remote cluster
+			current_block = LIST_REMOTE_FIRST( cxy, root , page_t , list );
+
+            // remove first free page from the free-list in remote cluster
+			list_remote_unlink( cxy , &current_block->list );
+
+            // register found block
+            found_block = current_block;
+
+	        // decrement relevant free-list number of items in remote cluster
+	        hal_remote_atomic_add( XPTR( cxy , &ppm->free_pages_nr[current_order] ), -1 );
+
+            // compute found block size
+	        current_size = (1 << current_order);
+
+			break;
+		}
+	}
+
+	if( current_block == NULL ) // return failure
+	{
+		// release lock protecting free lists
+		remote_busylock_release( lock_xp );
+
+#if DEBUG_PPM_ALLOC_PAGES
+cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_PPM_ALLOC_PAGES < cycle )
+printk("\n[%s] thread[%x,%x] cannot allocate %d page(s) in cluster %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 1<<order, cxy, cycle );
+#endif
+
+		return XPTR_NULL;
+	}
+
+	// split the found block in smaller sub-blocks if required
+	// and update the free-lists accordingly in remote cluster
+	while( current_order > order )
+	{
+        // update order, size, and local pointer for new free block
+		current_order --;
+		current_size >>= 1;
+		current_block = found_block + current_size;
+
+        // update new free block order field in remote cluster 
+		hal_remote_s32( XPTR( cxy , &current_block->order ) , current_order );
+
+        // get local pointer on the root of the relevant free_list in remote cluster  
+        list_entry_t * root = &ppm->free_pages_root[current_order];
+
+        // insert new free block in this free_list
+		list_remote_add_first( cxy , root, &current_block->list );
+
+        // update free-list number of items in remote cluster
+        hal_remote_atomic_add( XPTR(cxy , &ppm->free_pages_nr[current_order]), 1 );
+	}
+
+	// update refcount, flags and order fields in found block remote page descriptor
+	page_remote_clear_flag( XPTR( cxy , found_block ), PG_FREE );
+	page_remote_refcount_up( XPTR( cxy , found_block ) );
+	hal_remote_s32( XPTR( cxy , &found_block->order ) , order );
+    
+	// release lock protecting free lists in remote cluster
+	remote_busylock_release( lock_xp );
+
+    // update DQDT page counter in remote cluster
+    dqdt_increment_pages( cxy , order );
+
+#if DEBUG_PPM_ALLOC_PAGES
+cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_PPM_ALLOC_PAGES < cycle )
+printk("\n[%s] thread[%x,%x] exit for %d page(s) / ppn = %x in cluster %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 
+1<<order, ppm_page2ppn(XPTR( local_cxy , found_block )), cxy, cycle );
+#endif
+
+#if(DEBUG_PPM_ALLOC_PAGES & 0x1)
+if( DEBUG_PPM_ALLOC_PAGES < cycle )
+ppm_remote_display( cxy );
+#endif
+
+	return XPTR( cxy , found_block );
+
+}  // end ppm_remote_alloc_pages()
+
+//////////////////////////////////////////
+void ppm_remote_free_pages( cxy_t     cxy,
+                            page_t  * page )
+{
+    xptr_t     page_xp;          // extended pointer on released page descriptor
+    uint32_t   order;            // released block order
+	page_t   * buddy_ptr;        // searched buddy block page descriptor
+    uint32_t   buddy_order;      // searched buddy block order
+	uint32_t   buddy_index;      // buddy block index in page_tbl[]
+	page_t   * current_ptr;      // current (merged) block page descriptor
+	uint32_t   current_index;    // current (merged) block index in page_tbl[]
+	uint32_t   current_order;    // current (merged) block order
+
+#if DEBUG_PPM_FREE_PAGES
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_PPM_FREE_PAGES < cycle )
+printk("\n[%s] thread[%x,%x] enter for %d page(s) in cluster %x / ppn %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 
+1<<page->order, cxy, ppm_page2ppn(XPTR(cxy , page)), cycle );
+#endif
+
+#if(DEBUG_PPM_FREE_PAGES & 0x1)
+if( DEBUG_PPM_FREE_PAGES < cycle )
+ppm_remote_display( cxy );
+#endif
+
+    // build extended pointer on released page descriptor
+    page_xp = XPTR( cxy , page );
+    
+    // get released page order
+    order = hal_remote_l32( XPTR( cxy , &page->order ) );
+
+    // get local pointer on PPM (same in all clusters)
+	ppm_t * ppm = &LOCAL_CLUSTER->ppm;
+
+    // build extended pointer on lock protecting remote PPM
+    xptr_t lock_xp = XPTR( cxy , &ppm->free_lock );
+
+    // get local pointer on remote PPM page_tbl[] array 
+	page_t * pages_tbl = hal_remote_lpt( XPTR( cxy , &ppm->pages_tbl ) );
+
+	// get lock protecting free_pages in remote cluster
+	remote_busylock_acquire( lock_xp );
+
+assert( !page_remote_is_flag( page_xp , PG_FREE ) ,
+"page already released : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
+
+assert( !page_remote_is_flag( page_xp , PG_RESERVED ) ,
+"reserved page : ppn = %x\n" , ppm_page2ppn(XPTR(local_cxy,page)) );
+
+	// update released page descriptor flags
+	page_remote_set_flag( page_xp , PG_FREE );
+
+	// search the buddy page descriptor
+	// - merge with current page descriptor if found
+	// - exit to release the current page descriptor if not found
+	current_ptr   = page;
+	current_index = (uint32_t)(page - ppm->pages_tbl);
+	for( current_order = order ;
+	     current_order < CONFIG_PPM_MAX_ORDER ;
+	     current_order++ )
+	{
+		buddy_index = current_index ^ (1 << current_order);
+		buddy_ptr   = pages_tbl + buddy_index;
+
+        // get buddy block order
+        buddy_order = hal_remote_l32( XPTR( cxy , &buddy_ptr->order ) );
+
+        // exit loop if buddy block not found
+		if( !page_remote_is_flag( XPTR( cxy , buddy_ptr ) , PG_FREE ) || 
+            (buddy_order != current_order) ) break;
+
+		// remove buddy from free list in remote cluster
+		list_remote_unlink( cxy , &buddy_ptr->list );
+        hal_remote_atomic_add( XPTR( cxy , &ppm->free_pages_nr[current_order] ) , -1 );
+
+        // reset order field in buddy block page descriptor
+        hal_remote_s32( XPTR( cxy , &buddy_ptr->order ) , 0 );
+
+		// compute merged block index in page_tbl[] array 
+		current_index &= buddy_index;
+	}
+
+	// update merged page descriptor order field
+	current_ptr = pages_tbl + current_index;
+    hal_remote_s32( XPTR( cxy , &current_ptr->order ) , current_order );
+
+	// insert merged block into relevant free list in remote cluster
+	list_remote_add_first( cxy , &ppm->free_pages_root[current_order] , &current_ptr->list );
+    hal_remote_atomic_add( XPTR( cxy , &ppm->free_pages_nr[current_order] ) , 1 );
+
+	// release lock protecting free_pages[] array
+	remote_busylock_release( lock_xp );
+
+    // update DQDT
+    dqdt_decrement_pages( cxy , page->order );
+
+#if DEBUG_PPM_FREE_PAGES
+cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_PPM_FREE_PAGES < cycle )
+printk("\n[%s] thread[%x,%x] exit for %d page(s) in cluster %x / ppn %x / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 
+1<<page->order, cxy, ppm_page2ppn(XPTR(cxy , page)), cycle );
+#endif
+
+#if(DEBUG_PPM_FREE_PAGES & 0x1)
+if( DEBUG_PPM_FREE_PAGES < cycle )
+ppm_remote_display( cxy );
+#endif
+
+}  // end ppm_remote_free_pages()
+
+////////////////////////////////////
+void ppm_remote_display( cxy_t cxy )
 {
 	uint32_t       order;
@@ -358,6 +618,9 @@
     ppm_t * ppm = &LOCAL_CLUSTER->ppm;
 
-	// get lock protecting free lists
-	busylock_acquire( &ppm->free_lock );
+    // build extended pointer on lock protecting remote PPM
+    xptr_t lock_xp = XPTR( cxy , &ppm->free_lock );
+
+	// get lock protecting free lists in remote cluster
+	remote_busylock_acquire( lock_xp );
 
 	printk("\n***** PPM in cluster %x / %d pages\n", local_cxy , ppm->pages_nr );
@@ -365,8 +628,9 @@
 	for( order = 0 ; order < CONFIG_PPM_MAX_ORDER ; order++ )
 	{
-		printk("- order = %d / free_pages = %d\t: ",
-		       order , ppm->free_pages_nr[order] );
-
-		LIST_FOREACH( &ppm->free_pages_root[order] , iter )
+        // get number of free pages for free_list[order] in remote cluster
+        uint32_t n = hal_remote_l32( XPTR( cxy , &ppm->free_pages_nr[order] ) );
+		printk("- order = %d / free_pages = %d\t: ", order , n );
+
+		LIST_REMOTE_FOREACH( cxy , &ppm->free_pages_root[order] , iter )
 		{
 			page = LIST_ELEMENT( iter , page_t , list );
@@ -377,14 +641,16 @@
 	}
 
-	// release lock protecting free lists
-	busylock_release( &ppm->free_lock );
+	// release lock protecting free lists in remote cluster
+	remote_busylock_release( lock_xp );
 }
 
-///////////////////////////////////////
-error_t ppm_assert_order( ppm_t * ppm )
+////////////////////////////////
+error_t ppm_assert_order( void )
 {
 	uint32_t       order;
 	list_entry_t * iter;
 	page_t       * page;
+
+    ppm_t * ppm = &LOCAL_CLUSTER->ppm;
 
 	for( order=0 ; order < CONFIG_PPM_MAX_ORDER ; order++ )
@@ -438,33 +704,6 @@
         hal_remote_s32( page_flags_xp , flags | PG_DIRTY );
 
-		// The PPM dirty list is a LOCAL list !!!
-        // We must update 4 pointers to insert a new page in this list.
-        // We can use the standard LIST API when the page is local,
-        // but we cannot use the standard API if the page is remote...
-
-        if( page_cxy == local_cxy )         // locally update the PPM dirty list
-        {
-            list_add_first( &ppm->dirty_root , &page_ptr->list );
-        } 
-        else                                // remotely update the PPM dirty list
-        {
-            // get local and remote pointers on "root" list entry 
-            list_entry_t * root    = &ppm->dirty_root;
-            xptr_t         root_xp = XPTR( page_cxy , root );
-
-            // get local and remote pointers on "page" list entry 
-            list_entry_t * list    = &page_ptr->list;
-            xptr_t         list_xp = XPTR( page_cxy , list );
-
-            // get local and remote pointers on first dirty page 
-            list_entry_t * dirt    = hal_remote_lpt( XPTR( page_cxy, &root->next ) );
-            xptr_t         dirt_xp = XPTR( page_cxy , dirt );
-
-            // set root.next, list.next, list pred, curr.pred in remote cluster
-            hal_remote_spt( root_xp                    , list );
-            hal_remote_spt( list_xp                    , dirt );
-            hal_remote_spt( list_xp + sizeof(intptr_t) , root );
-            hal_remote_spt( dirt_xp + sizeof(intptr_t) , list );
-        }
+		// insert the page in the remote dirty list
+        list_remote_add_first( page_cxy , &ppm->dirty_root , &page_ptr->list );
 
 		done = true;
@@ -512,33 +751,6 @@
         hal_remote_s32( page_flags_xp , flags & (~PG_DIRTY) );
 
-		// The PPM dirty list is a LOCAL list !!!
-        // We must update 4 pointers to remove a page from this list.
-        // we can use the standard LIST API when the page is local,
-        // but we cannot use the standard API if the page is remote...
-
-        if( page_cxy == local_cxy )         // locally update the PPM dirty list
-        {
-            list_unlink( &page_ptr->list );
-        } 
-        else                                // remotely update the PPM dirty list
-        {
-            // get local and remote pointers on "page" list entry 
-            list_entry_t * list    = &page_ptr->list;
-            xptr_t         list_xp = XPTR( page_cxy , list );
-
-            // get local and remote pointers on "next" page list entry 
-            list_entry_t * next    = hal_remote_lpt( list_xp );
-            xptr_t         next_xp = XPTR( page_cxy , next );
-
-            // get local and remote pointers on "pred" page list entry 
-            list_entry_t * pred    = hal_remote_lpt( list_xp + sizeof(intptr_t) );
-            xptr_t         pred_xp = XPTR( page_cxy , pred );
-
-            // set root.next, list.next, list pred, curr.pred in remote cluster
-            hal_remote_spt( pred_xp                    , next );
-            hal_remote_spt( list_xp                    , NULL );
-            hal_remote_spt( list_xp + sizeof(intptr_t) , NULL );
-            hal_remote_spt( next_xp + sizeof(intptr_t) , pred );
-        }
+        // remove the page from remote dirty list
+        list_remote_unlink( page_cxy , &page_ptr->list );
 
 		done = true;
