Index: trunk/kernel/mm/kcm.c
===================================================================
--- trunk/kernel/mm/kcm.c	(revision 619)
+++ trunk/kernel/mm/kcm.c	(revision 635)
@@ -1,7 +1,6 @@
 /*
- * kcm.c - Per cluster Kernel Cache Manager implementation.
+ * kcm.c -  Kernel Cache Manager implementation.
  *
- * Author  Ghassan Almaless (2008,2009,2010,2011,2012)
- *         Alain Greiner    (2016,2017,2018,2019)
+ * Author  Alain Greiner    (2016,2017,2018,2019)
  *
  * Copyright (c) UPMC Sorbonne Universites
@@ -38,224 +37,212 @@
 
 
+/////////////////////////////////////////////////////////////////////////////////////
+//        Local access functions
+/////////////////////////////////////////////////////////////////////////////////////
+
 //////////////////////////////////////////////////////////////////////////////////////
-// This static function returns pointer on an allocated block from an active page.
-// It returns NULL if no block available in selected page.
-// It changes the page status if required.
+// This static function must be called by a local thread.
+// It returns a pointer on a block allocated from a non-full kcm_page.
+// It makes a panic if no block is available in selected page.
+// It changes the page status as required.
 //////////////////////////////////////////////////////////////////////////////////////
-// @ kcm      : pointer on kcm allocator.
-// @ kcm_page : pointer on active kcm page to use.
-/////////////////////////////////////////////////////////////////////////////////////
-static void * kcm_get_block( kcm_t      * kcm,
-                             kcm_page_t * kcm_page )
-{
-
-#if DEBUG_KCM
-thread_t * this = CURRENT_THREAD;
-uint32_t cycle = (uint32_t)hal_get_cycles();
+// @ kcm      : pointer on KCM allocator.
+// @ kcm_page : pointer on a non-full kcm_page.
+// @ return pointer on allocated block.
+/////////////////////////////////////////////////////////////////////////////////////
+static void * __attribute__((noinline)) kcm_get_block( kcm_t      * kcm,
+                                                       kcm_page_t * kcm_page )
+{
+    // initialise variables
+    uint32_t size   = 1 << kcm->order;
+    uint32_t max    = kcm->max_blocks;
+    uint32_t count  = kcm_page->count;
+    uint64_t status = kcm_page->status;
+
+assert( (count < max) , "kcm_page should not be full" );
+
+    uint32_t index  = 1;
+    uint64_t mask   = (uint64_t)0x2;
+    uint32_t found  = 0;
+
+	// allocate first free block in kcm_page, update status,
+    // and count , compute index of allocated block in kcm_page 
+    while( index <= max )
+    {
+        if( (status & mask) == 0 )   // block non allocated
+        {
+            kcm_page->status = status | mask;
+            kcm_page->count  = count + 1;
+            found  = 1;
+
+            break;      
+        }
+        
+        index++;
+        mask <<= 1;
+    }
+
+    // change the page list if almost full
+    if( count == max-1 )
+    {
+		list_unlink( &kcm_page->list);
+		kcm->active_pages_nr--;
+
+        list_add_first( &kcm->full_root , &kcm_page->list );
+		kcm->full_pages_nr ++;
+    }
+
+	// compute return pointer
+	void * ptr = (void *)((intptr_t)kcm_page + (index * size) );
+
+#if (DEBUG_KCM & 1)
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_KCM < cycle )
-printk("\n[%s] thread[%x,%x] enters for %s / page %x / count %d / active %d\n",
-__FUNCTION__, this->process->pid, this->trdid, kmem_type_str(kcm->type),
-(intptr_t)kcm_page , kcm_page->count , kcm_page->active );
-#endif
-
-assert( kcm_page->active , "kcm_page should be active" );
-
-	// get first block available
-	int32_t index = bitmap_ffs( kcm_page->bitmap , kcm->blocks_nr );
-
-assert( (index != -1) , "kcm_page should not be full" );
-
-	// allocate block
-	bitmap_clear( kcm_page->bitmap , index );
-
-	// increase kcm_page count
-	kcm_page->count ++;
-
-	// change the kcm_page to busy if no more free block in page
-	if( kcm_page->count >= kcm->blocks_nr )
-	{
-		kcm_page->active = 0;
-		list_unlink( &kcm_page->list);
-		kcm->active_pages_nr --;
-
-		list_add_first( &kcm->busy_root , &kcm_page->list);
-		kcm->busy_pages_nr ++;
-		kcm_page->busy = 1;
-	}
-
-	// compute return pointer
-	void * ptr = (void *)((intptr_t)kcm_page + CONFIG_KCM_SLOT_SIZE
-	             + (index * kcm->block_size) );
-
-#if DEBUG_KCM
-cycle = (uint32_t)hal_get_cycles();
-if( DEBUG_KCM < cycle )
-printk("\n[%s] thread[%x,%x] exit for %s / ptr %x / page %x / count %d\n",
-__FUNCTION__, this->process->pid, this->trdid, kmem_type_str(kcm->type),
-(intptr_t)ptr, (intptr_t)kcm_page, kcm_page->count );
+printk("\n[%s] thread[%x,%x] allocated block %x in page %x / size %d / count %d / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, ptr, kcm_page, size, count + 1, cycle );
 #endif
 
 	return ptr;
-}
-
-/////////////////////////////////////////////////////////////////////////////////////
-// This static function releases a previously allocated block.
-// It changes the kcm_page status if required.
-/////////////////////////////////////////////////////////////////////////////////////
-// @ kcm      : pointer on kcm allocator.
-// @ kcm_page : pointer on kcm_page.
-// @ ptr      : pointer on block to be released.
-/////////////////////////////////////////////////////////////////////////////////////
-static void kcm_put_block ( kcm_t      * kcm,
-                            kcm_page_t * kcm_page,
-                            void       * ptr )
-{
-	uint32_t     index;
-
+
+}  // end kcm_get_block()
+
+/////////////////////////////////////////////////////////////////////////////////////
+// This private static function must be called by a local thread.
+// It releases a previously allocated block to the relevant kcm_page.
+// It makes a panic if the released block is not allocated in this page.
+// It changes the kcm_page status as required.
+/////////////////////////////////////////////////////////////////////////////////////
+// @ kcm        : pointer on kcm allocator.
+// @ kcm_page   : pointer on kcm_page.
+// @ block_ptr  : pointer on block to be released.
+/////////////////////////////////////////////////////////////////////////////////////
+static void __attribute__((noinline)) kcm_put_block ( kcm_t      * kcm,
+                                                      kcm_page_t * kcm_page,
+                                                      void       * block_ptr )
+{
+    // initialise variables
+    uint32_t max    = kcm->max_blocks;
+    uint32_t size   = 1 << kcm->order;
+    uint32_t count  = kcm_page->count;
+    uint64_t status = kcm_page->status;
+    
 	// compute block index from block pointer
-	index = ((uint8_t *)ptr - (uint8_t *)kcm_page - CONFIG_KCM_SLOT_SIZE) / kcm->block_size;
-
-assert( !bitmap_state( kcm_page->bitmap , index ) , "page already freed" );
-
-assert( (kcm_page->count > 0) , "count already zero" );
-
-	bitmap_set( kcm_page->bitmap , index );
-	kcm_page->count --;
-
-	// change the page to active if it was busy
-	if( kcm_page->busy )
-	{
-		kcm_page->busy = 0;
+	uint32_t index = ((intptr_t)block_ptr - (intptr_t)kcm_page) / size;
+
+    // compute mask in bit vector
+    uint64_t mask = ((uint64_t)0x1) << index;
+
+assert( (status & mask) , "released block not allocated : status (%x,%x) / mask(%x,%x)",
+GET_CXY(status), GET_PTR(status), GET_CXY(mask  ), GET_PTR(mask  ) );
+
+    // update status & count in kcm_page 
+	kcm_page->status = status & ~mask;
+	kcm_page->count  = count - 1;
+
+	// change the page mode if page was full
+	if( count == max )
+	{
 		list_unlink( &kcm_page->list );
-		kcm->busy_pages_nr --;
+		kcm->full_pages_nr --;
 
 		list_add_last( &kcm->active_root, &kcm_page->list );
 		kcm->active_pages_nr ++;
-		kcm_page->active = 1;
-	}
-
-	// change the kcm_page to free if last block in active page
-	if( (kcm_page->active) && (kcm_page->count == 0) )
-	{
-		kcm_page->active = 0;
-		list_unlink( &kcm_page->list);
-		kcm->active_pages_nr --;
-
-		list_add_first( &kcm->free_root , &kcm_page->list);
-		kcm->free_pages_nr ++;
-	}
-}
-
-/////////////////////////////////////////////////////////////////////////////////////
-// This static function allocates one page from PPM. It initializes
-// the kcm_page descriptor, and introduces the new kcm_page into freelist.
-/////////////////////////////////////////////////////////////////////////////////////
-static error_t freelist_populate( kcm_t * kcm )
-{
-	page_t     * page;
-	kcm_page_t * kcm_page;
-	kmem_req_t   req;
-
-	// get one page from local PPM
-	req.type  = KMEM_PAGE;
-	req.size  = 0;
-	req.flags = AF_KERNEL;
-	page = kmem_alloc( &req );
-
-	if( page == NULL )
-	{
-		printk("\n[ERROR] in %s : failed to allocate page in cluster %x\n",
-	    __FUNCTION__ , local_cxy );
-		return ENOMEM;
-	}
-
-	// get page base address
-	xptr_t base_xp = ppm_page2base( XPTR( local_cxy , page ) );
-	kcm_page = (kcm_page_t *)GET_PTR( base_xp );
-
-	// initialize KCM-page descriptor
-	bitmap_set_range( kcm_page->bitmap , 0 , kcm->blocks_nr );
-
-	kcm_page->busy          = 0;
-	kcm_page->active        = 0;
-	kcm_page->count      = 0;
-	kcm_page->kcm           = kcm;
-	kcm_page->page          = page;
-
-	// introduce new page in free-list
-	list_add_first( &kcm->free_root , &kcm_page->list );
-	kcm->free_pages_nr ++;
-
-	return 0;
-}
-
-/////////////////////////////////////////////////////////////////////////////////////
-// This private function gets one KCM page from the KCM freelist.
-// It populates the freelist if required.
-/////////////////////////////////////////////////////////////////////////////////////
-static kcm_page_t * freelist_get( kcm_t * kcm )
-{
-	error_t      error;
-	kcm_page_t * kcm_page;
-
-	// get a new page from PPM if freelist empty
-	if( kcm->free_pages_nr == 0 )
-	{
-		error = freelist_populate( kcm );
-		if( error ) return NULL;
-	}
-
-	// get first KCM page from freelist and unlink it
-	kcm_page = LIST_FIRST( &kcm->free_root, kcm_page_t , list );
-	list_unlink( &kcm_page->list );
-	kcm->free_pages_nr --;
+	}
+
+#if (DEBUG_KCM & 1)
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM < cycle )
+printk("\n[%s] thread[%x,%x] released block %x in page %x / size %d / count %d / cycle %d\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, kcm_page, size, count - 1, cycle );
+#endif
+
+}  // kcm_put_block()
+
+/////////////////////////////////////////////////////////////////////////////////////
+// This private static function must be called by a local thread.
+// It returns one non-full kcm_page with te following policy :
+// - if the "active_list" is non empty, it returns the first "active" page,
+//   without modifying the KCM state.
+// - if the "active_list" is empty, it allocates a new page fromm PPM, inserts
+//   this page in the active_list, and returns it. 
+/////////////////////////////////////////////////////////////////////////////////////
+// @ kcm      : local pointer on local KCM allocator.
+// @ return pointer on a non-full kcm page if success / returns NULL if no memory.
+/////////////////////////////////////////////////////////////////////////////////////
+static kcm_page_t * __attribute__((noinline)) kcm_get_page( kcm_t * kcm )
+{
+    kcm_page_t * kcm_page;
+
+    uint32_t active_pages_nr = kcm->active_pages_nr;
+
+    if( active_pages_nr > 0 )       // return first active page
+    {
+        kcm_page = LIST_FIRST( &kcm->active_root , kcm_page_t , list );
+    }
+    else                            // allocate a new page from PPM
+	{
+        // get one 4 Kbytes page from local PPM
+        page_t * page = ppm_alloc_pages( 0 );
+
+	    if( page == NULL )
+	    {
+		    printk("\n[ERROR] in %s : failed to allocate page in cluster %x\n",
+	        __FUNCTION__ , local_cxy );
+
+		    return NULL;
+        }
+
+	    // get page base address
+	    xptr_t base_xp = ppm_page2base( XPTR( local_cxy , page ) );
+
+        // get local pointer on kcm_page
+	    kcm_page = GET_PTR( base_xp );
+
+	    // initialize kcm_page descriptor
+	    kcm_page->status = 0;
+	    kcm_page->count  = 0;
+	    kcm_page->kcm    = kcm;
+	    kcm_page->page   = page;
+
+	    // introduce new page in KCM active_list
+	    list_add_first( &kcm->active_root , &kcm_page->list );
+	    kcm->active_pages_nr ++;
+	}
 
 	return kcm_page;
-}
+
+}  // end kcm_get_page()
 
 //////////////////////////////
 void kcm_init( kcm_t    * kcm,
-	           uint32_t   type )
-{
-
-// the kcm_page descriptor must fit in the KCM slot
-assert( (sizeof(kcm_page_t) <= CONFIG_KCM_SLOT_SIZE) , "KCM slot too small\n" );
-
-// the allocated object must fit in one single page
-assert( (kmem_type_size(type) <= (CONFIG_PPM_PAGE_SIZE - CONFIG_KCM_SLOT_SIZE)),
-"allocated object requires more than one single page\n" );
+	           uint32_t   order)
+{
+
+assert( ((order > 5) && (order < 12)) , "order must be in [6,11]" );
 
 	// initialize lock
-	busylock_init( &kcm->lock , LOCK_KCM_STATE );
-
-	// initialize KCM type
-	kcm->type = type;
+	remote_busylock_init( XPTR( local_cxy , &kcm->lock ) , LOCK_KCM_STATE );
 
 	// initialize KCM page lists
-	kcm->free_pages_nr   = 0;
-	kcm->busy_pages_nr   = 0;
+	kcm->full_pages_nr   = 0;
 	kcm->active_pages_nr = 0;
-	list_root_init( &kcm->free_root );
-	list_root_init( &kcm->busy_root );
+	list_root_init( &kcm->full_root );
 	list_root_init( &kcm->active_root );
 
-	// initialize block size
-	uint32_t block_size = ARROUND_UP( kmem_type_size( type ) , CONFIG_KCM_SLOT_SIZE );
-	kcm->block_size = block_size;
-
-	// initialize number of blocks per page
-	uint32_t  blocks_nr = (CONFIG_PPM_PAGE_SIZE - CONFIG_KCM_SLOT_SIZE) / block_size;
-	kcm->blocks_nr = blocks_nr;
-
+	// initialize order and max_blocks
+	kcm->order      = order;
+    kcm->max_blocks = ( CONFIG_PPM_PAGE_SIZE >> order ) - 1;
+ 
 #if DEBUG_KCM
 thread_t * this  = CURRENT_THREAD;
 uint32_t   cycle = (uint32_t)hal_get_cycles();
 if( DEBUG_KCM < cycle )
-printk("\n[%s] thread[%x,%x] initialised KCM %s : block_size %d / blocks_nr %d\n",
-__FUNCTION__, this->process->pid, this->trdid,
-kmem_type_str( kcm->type ), block_size, blocks_nr );
-#endif
-
-}
+printk("\n[%s] thread[%x,%x] initialised KCM / order %d / max_blocks %d\n",
+__FUNCTION__, this->process->pid, this->trdid, order, kcm->max_blocks );
+#endif
+
+}  // end kcm_init()
 
 ///////////////////////////////
@@ -263,111 +250,390 @@
 {
 	kcm_page_t   * kcm_page;
-	list_entry_t * iter;
+
+    // build extended pointer on  KCM lock
+    xptr_t lock_xp = XPTR( local_cxy , &kcm->lock );
 
 	// get KCM lock
-	busylock_acquire( &kcm->lock );
-
-	// release all free pages
-	LIST_FOREACH( &kcm->free_root , iter )
-	{
-		kcm_page = (kcm_page_t *)LIST_ELEMENT( iter , kcm_page_t , list );
-		list_unlink( iter );
-		kcm->free_pages_nr --;
+	remote_busylock_acquire( lock_xp );
+
+	// release all full pages
+	while( list_is_empty( &kcm->full_root ) == false )
+	{
+		kcm_page = LIST_FIRST( &kcm->full_root , kcm_page_t , list );
+		list_unlink( &kcm_page->list );
 		ppm_free_pages( kcm_page->page );
 	}
 
-	// release all active pages
-	LIST_FOREACH( &kcm->active_root , iter )
-	{
-		kcm_page = (kcm_page_t *)LIST_ELEMENT( iter , kcm_page_t , list );
-		list_unlink( iter );
-		kcm->free_pages_nr --;
+    // release all empty pages
+    while( list_is_empty( &kcm->active_root ) == false )
+	{
+		kcm_page = LIST_FIRST( &kcm->active_root , kcm_page_t , list );
+		list_unlink( &kcm_page->list );
 		ppm_free_pages( kcm_page->page );
 	}
 
-	// release all busy pages
-	LIST_FOREACH( &kcm->busy_root , iter )
-	{
-		kcm_page = (kcm_page_t *)LIST_ELEMENT( iter , kcm_page_t , list );
-		list_unlink( iter );
-		kcm->free_pages_nr --;
-		ppm_free_pages( kcm_page->page );
-	}
-
 	// release KCM lock
-	busylock_release( &kcm->lock );
+	remote_busylock_release( lock_xp );
 }
 
-///////////////////////////////
-void * kcm_alloc( kcm_t * kcm )
-{
+//////////////////////////////////
+void * kcm_alloc( uint32_t order )
+{
+    kcm_t      * kcm_ptr;
 	kcm_page_t * kcm_page;
-	void       * ptr = NULL;   // pointer on block
+	void       * block_ptr;
+
+    // min block size is 64 bytes
+    if( order < 6 ) order = 6;
+
+assert( (order < 12) , "order = %d / must be less than 12" , order );
+
+    // get local pointer on relevant KCM allocator
+    kcm_ptr = &LOCAL_CLUSTER->kcm[order - 6];
+
+    // build extended pointer on local KCM lock
+    xptr_t lock_xp = XPTR( local_cxy , &kcm_ptr->lock );
+
+	// get KCM lock
+	remote_busylock_acquire( lock_xp );
+
+    // get a non-full kcm_page
+    kcm_page = kcm_get_page( kcm_ptr );
+
+    if( kcm_page == NULL )
+	{
+		remote_busylock_release( lock_xp );
+		return NULL;
+	}
+
+	// get a block from selected active page
+	block_ptr = kcm_get_block( kcm_ptr , kcm_page );
+
+	// release lock
+	remote_busylock_release( lock_xp );
+
+#if DEBUG_KCM
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM < cycle )
+printk("\n[%s] thread[%x,%x] allocated block %x / order %d / kcm %x / status[%x,%x] / count %d\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, order, kcm_ptr,
+GET_CXY(kcm_page->status), GET_PTR(kcm_page->status), kcm_page->count );
+#endif
+
+	return block_ptr;
+
+}  // end kcm_alloc()
+
+/////////////////////////////////
+void kcm_free( void * block_ptr )
+{
+    kcm_t      * kcm_ptr;
+	kcm_page_t * kcm_page;
+
+// check argument
+assert( (block_ptr != NULL) , "block pointer cannot be NULL" );
+
+    // get local pointer on KCM page
+	kcm_page = (kcm_page_t *)((intptr_t)block_ptr & ~CONFIG_PPM_PAGE_MASK);
+
+    // get local pointer on KCM descriptor
+	kcm_ptr = kcm_page->kcm;
+
+#if DEBUG_KCM
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM < cycle )
+printk("\n[%s] thread[%x,%x] release block %x / order %d / kcm %x / status [%x,%x] / count %d\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, kcm_ptr->order, kcm_ptr,
+GET_CXY(kcm_page->status), GET_PTR(kcm_page->status), kcm_page->count );
+#endif
+
+    // build extended pointer on local KCM lock
+    xptr_t lock_xp = XPTR( local_cxy , &kcm_ptr->lock );
 
 	// get lock
-	busylock_acquire( &kcm->lock );
-
-	// get an active page
-	if( list_is_empty( &kcm->active_root ) )  // no active page => get one
-	{
-		// get a page from free list
-		kcm_page = freelist_get( kcm );
-
-		if( kcm_page == NULL )
-		{
-			busylock_release( &kcm->lock );
-			return NULL;
-		}
-
-		// insert page in active list
-		list_add_first( &kcm->active_root , &kcm_page->list );
-		kcm->active_pages_nr ++;
-		kcm_page->active = 1;
-	}
-	else                                    // get first page from active list
-	{
-		// get page pointer from active list
-		kcm_page = (kcm_page_t *)LIST_FIRST( &kcm->active_root , kcm_page_t , list );
+	remote_busylock_acquire( lock_xp );
+
+	// release block
+	kcm_put_block( kcm_ptr , kcm_page , block_ptr );
+
+	// release lock
+	remote_busylock_release( lock_xp );
+}
+
+/////////////////////////////////////////////////////////////////////////////////////
+//        Remote access functions
+/////////////////////////////////////////////////////////////////////////////////////
+
+/////////////////////////////////////////////////////////////////////////////////////
+// This static function can be called by any thread running in any cluster.
+// It returns a local pointer on a block allocated from an non-full kcm_page.
+// It makes a panic if no block available in selected page.
+// It changes the page status as required.
+/////////////////////////////////////////////////////////////////////////////////////
+// @ kcm_cxy  : remote KCM cluster identidfier.
+// @ kcm_ptr  : local pointer on remote KCM allocator.
+// @ kcm_page : pointer on active kcm page to use.
+// @ return a local pointer on the allocated block. 
+/////////////////////////////////////////////////////////////////////////////////////
+static void * __attribute__((noinline)) kcm_remote_get_block( cxy_t        kcm_cxy,
+                                                              kcm_t      * kcm_ptr,
+                                                              kcm_page_t * kcm_page )
+{
+    uint32_t order  = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->order ) );
+    uint32_t max    = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->max_blocks ) );
+    uint32_t count  = hal_remote_l32( XPTR( kcm_cxy , &kcm_page->count ) );
+    uint64_t status = hal_remote_l64( XPTR( kcm_cxy , &kcm_page->status ) );
+    uint32_t size   = 1 << order;
+
+assert( (count < max) , "kcm_page should not be full" );
+
+    uint32_t index  = 1;
+    uint64_t mask   = (uint64_t)0x2;
+    uint32_t found  = 0;
+    
+	// allocate first free block in kcm_page, update status,
+    // and count , compute index of allocated block in kcm_page 
+    while( index <= max )
+    {
+        if( (status & mask) == 0 )   // block non allocated
+        {
+            hal_remote_s64( XPTR( kcm_cxy , &kcm_page->status ) , status | mask );
+            hal_remote_s64( XPTR( kcm_cxy , &kcm_page->count  ) , count + 1 );
+            found  = 1; 
+            break;      
+        }
+        
+        index++;
+        mask <<= 1;
+    }
+
+	// change the page list if almost full
+	if( count == max-1 )
+	{
+		list_remote_unlink( kcm_cxy , &kcm_page->list );
+		hal_remote_atomic_add( XPTR( kcm_cxy , &kcm_ptr->active_pages_nr ) , -1 );
+
+		list_remote_add_first( kcm_cxy , &kcm_ptr->full_root , &kcm_page->list );
+		hal_remote_atomic_add( XPTR( kcm_cxy , &kcm_ptr->full_pages_nr ) , 1 );
+	}
+
+	// compute return pointer
+	void * ptr = (void *)((intptr_t)kcm_page + (index * size) );
+
+#if DEBUG_KCM_REMOTE
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM_REMOTE < cycle )
+printk("\n[%s] thread[%x,%x] get block %x in page %x / cluster %x / size %x / count %d\n",
+__FUNCTION__, this->process->pid, this->trdid, 
+ptr, kcm_page, kcm_cxy, size, count + 1 );
+#endif
+
+	return ptr;
+
+}  // end kcm_remote_get_block()
+
+/////////////////////////////////////////////////////////////////////////////////////
+// This private static function can be called by any thread running in any cluster.
+// It releases a previously allocated block to the relevant kcm_page.
+// It changes the kcm_page status as required.
+/////////////////////////////////////////////////////////////////////////////////////
+// @ kcm_cxy   : remote KCM cluster identifier
+// @ kcm_ptr   : local pointer on remote KCM.
+// @ kcm_page  : local pointer on kcm_page.
+// @ block_ptr : pointer on block to be released.
+/////////////////////////////////////////////////////////////////////////////////////
+static void __attribute__((noinline)) kcm_remote_put_block ( cxy_t        kcm_cxy,
+                                                             kcm_t      * kcm_ptr,
+                                                             kcm_page_t * kcm_page,
+                                                             void       * block_ptr )
+{
+    uint32_t max    = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->max_blocks ) );
+    uint32_t order  = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->order ) );
+    uint32_t count  = hal_remote_l32( XPTR( kcm_cxy , &kcm_page->count ) );
+    uint64_t status = hal_remote_l64( XPTR( kcm_cxy , &kcm_page->status ) );
+    uint32_t size   = 1 << order;
+    
+	// compute block index from block pointer
+	uint32_t index = ((intptr_t)block_ptr - (intptr_t)kcm_page) / size;
+
+    // compute mask in bit vector
+    uint64_t mask = 1 << index;
+
+assert( (status & mask) , "released page not allocated" );
+
+    // update status & count in kcm_page 
+	hal_remote_s64( XPTR( kcm_cxy , &kcm_page->status ) , status & ~mask );
+	hal_remote_s32( XPTR( kcm_cxy , &kcm_page->count  ) , count - 1 );
+
+	// change the page list if page was full
+	if( count == max )
+	{
+		list_remote_unlink( kcm_cxy , &kcm_page->list );
+		hal_remote_atomic_add( XPTR( kcm_cxy , &kcm_ptr->full_pages_nr ) , -1 );
+
+		list_remote_add_last( kcm_cxy , &kcm_ptr->active_root, &kcm_page->list );
+		hal_remote_atomic_add( XPTR( kcm_cxy , &kcm_ptr->active_pages_nr ) , 1 );
+	}
+
+#if (DEBUG_KCM_REMOTE & 1)
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM_REMOTE < cycle )
+printk("\n[%s] thread[%x,%x] released block %x in page %x / cluster %x / size %x / count %d\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, kcm_page, size, count - 1 )
+#endif
+
+}  // end kcm_remote_put_block()
+
+/////////////////////////////////////////////////////////////////////////////////////
+// This private static function can be called by any thread running in any cluster.
+// It gets one non-full KCM page from the remote KCM.
+// It allocates a page from remote PPM to populate the freelist, and initialises
+// the kcm_page descriptor when required.
+/////////////////////////////////////////////////////////////////////////////////////
+static kcm_page_t * __attribute__((noinline)) kcm_remote_get_page( cxy_t    kcm_cxy,
+                                                                   kcm_t  * kcm_ptr )
+{
+    kcm_page_t * kcm_page;    // local pointer on remote KCM page 
+
+    uint32_t active_pages_nr = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->active_pages_nr ) );
+
+    if( active_pages_nr > 0 )       // return first active page
+    {
+        kcm_page = LIST_REMOTE_FIRST( kcm_cxy , &kcm_ptr->active_root , kcm_page_t , list );
+    }
+    else                            // allocate a new page from PPM
+	{
+        // get one 4 Kbytes page from remote PPM
+        page_t * page = ppm_remote_alloc_pages( kcm_cxy , 0 );
+
+	    if( page == NULL )
+	    {
+		    printk("\n[ERROR] in %s : failed to allocate page in cluster %x\n",
+	        __FUNCTION__ , kcm_cxy );
+
+		    return NULL;
+        }
+
+	    // get remote page base address
+	    xptr_t base_xp = ppm_page2base( XPTR( kcm_cxy , page ) );
+
+        // get local pointer on kcm_page
+	    kcm_page = GET_PTR( base_xp );
+
+	    // initialize kcm_page descriptor
+	    hal_remote_s32( XPTR( kcm_cxy , &kcm_page->count )  , 0 );
+	    hal_remote_s64( XPTR( kcm_cxy , &kcm_page->status ) , 0 );
+	    hal_remote_spt( XPTR( kcm_cxy , &kcm_page->kcm )    , kcm_ptr );
+	    hal_remote_spt( XPTR( kcm_cxy , &kcm_page->page )   , page );
+
+	    // introduce new page in remote KCM active_list
+	    list_remote_add_first( kcm_cxy , &kcm_ptr->active_root , &kcm_page->list );
+	    hal_remote_atomic_add( XPTR( kcm_cxy , &kcm_ptr->active_pages_nr ) , 1 );
+	}
+
+	return kcm_page;
+
+}  // end kcm_remote_get_page()
+
+/////////////////////////////////////////
+void * kcm_remote_alloc( cxy_t    kcm_cxy,
+                         uint32_t order )
+{
+    kcm_t      * kcm_ptr;
+    kcm_page_t * kcm_page;
+    void       * block_ptr;
+
+    if( order < 6 ) order = 6;
+
+assert( (order < 12) , "order = %d / must be less than 12" , order );
+
+    // get local pointer on relevant KCM allocator
+    kcm_ptr = &LOCAL_CLUSTER->kcm[order - 6];
+
+    // build extended pointer on remote KCM lock
+    xptr_t lock_xp = XPTR( kcm_cxy , &kcm_ptr->lock );
+
+	// get lock
+	remote_busylock_acquire( lock_xp );
+
+    // get a non-full kcm_page
+    kcm_page = kcm_remote_get_page( kcm_cxy , kcm_ptr );
+
+    if( kcm_page == NULL )
+	{
+		remote_busylock_release( lock_xp );
+		return NULL;
 	}
 
 	// get a block from selected active page
-	// cannot fail, as an active page cannot be full...
-	ptr  = kcm_get_block( kcm , kcm_page );
+	block_ptr = kcm_remote_get_block( kcm_cxy , kcm_ptr , kcm_page );
 
 	// release lock
-	busylock_release( &kcm->lock );
-
-	return ptr;
+	remote_busylock_release( lock_xp );
+
+#if DEBUG_KCM_REMOTE
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+if( DEBUG_KCM_REMOTE < cycle )
+printk("\n[%s] thread[%x,%x] allocated block %x / order %d / kcm[%x,%x]\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, order, kcm_cxy, kcm_ptr );
+#endif
+
+	return block_ptr;
+
+}  // end kcm_remote_alloc()
+
+/////////////////////////////////////
+void kcm_remote_free( cxy_t  kcm_cxy,
+                      void * block_ptr )
+{
+	kcm_t      * kcm_ptr;
+	kcm_page_t * kcm_page;
+
+// check argument
+assert( (block_ptr != NULL) , "block pointer cannot be NULL" );
+
+    // get local pointer on remote KCM page
+	kcm_page = (kcm_page_t *)((intptr_t)block_ptr & ~CONFIG_PPM_PAGE_MASK);
+
+    // get local pointer on remote KCM
+	kcm_ptr = hal_remote_lpt( XPTR( kcm_cxy , &kcm_page->kcm ) );
+
+    // build extended pointer on remote KCM lock
+    xptr_t lock_xp = XPTR( kcm_cxy , &kcm_ptr->lock );
+
+	// get lock
+	remote_busylock_acquire( lock_xp );
+
+	// release block
+	kcm_remote_put_block( kcm_cxy , kcm_ptr , kcm_page , block_ptr );
+
+	// release lock
+	remote_busylock_release( lock_xp );
+
+#if DEBUG_KCM_REMOTE
+thread_t * this  = CURRENT_THREAD;
+uint32_t   cycle = (uint32_t)hal_get_cycles();
+uint32_t   order = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->order ) );
+if( DEBUG_KCM_REMOTE < cycle )
+printk("\n[%s] thread[%x,%x] released block %x / order %d / kcm[%x,%x]\n",
+__FUNCTION__, this->process->pid, this->trdid, block_ptr, order, kcm_cxy, kcm_ptr );
+#endif
+
+}  // end kcm_remote_free
+
+/////////////////////////////////////////
+void kcm_remote_display( cxy_t   kcm_cxy,
+                         kcm_t * kcm_ptr )
+{
+    uint32_t order           = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->order) );
+    uint32_t full_pages_nr   = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->full_pages_nr ) );
+    uint32_t active_pages_nr = hal_remote_l32( XPTR( kcm_cxy , &kcm_ptr->active_pages_nr ) );
+
+	printk("*** KCM / cxy %x / order %d / full_pages %d / empty_pages %d / active_pages %d\n",
+	kcm_cxy, order, full_pages_nr, active_pages_nr );
 }
-
-///////////////////////////
-void kcm_free( void * ptr )
-{
-	kcm_page_t * kcm_page;
-	kcm_t      * kcm;
-
-// check argument
-assert( (ptr != NULL) , "pointer cannot be NULL" );
-
-	kcm_page = (kcm_page_t *)((intptr_t)ptr & ~CONFIG_PPM_PAGE_MASK);
-	kcm      = kcm_page->kcm;
-
-	// get lock
-	busylock_acquire( &kcm->lock );
-
-	// release block
-	kcm_put_block( kcm , kcm_page , ptr );
-
-	// release lock
-	busylock_release( &kcm->lock );
-}
-
-////////////////////////////
-void kcm_print (kcm_t * kcm)
-{
-	printk("*** KCM type = %s / free_pages = %d / busy_pages = %d / active_pages = %d\n",
-	       kmem_type_str( kcm->type ) ,
-	       kcm->free_pages_nr ,
-	       kcm->busy_pages_nr ,
-	       kcm->active_pages_nr );
-}
