Re: [Hibernate] DistributedCacheConcurrencyStrategy implemented !

[Gavin K. <ga...@ap...>]

Sorry for not following this closely today. I am in the middle of a massive
(and long overdue) refactor of Hibernate (that does not affect the cache
package). I will pay closer attention to this once I am done with the
refactor.

peace...

----- Original Message -----
From: "Christian Meunier" <vc...@cl...>
To: <hib...@li...>
Sent: Tuesday, September 24, 2002 12:38 AM
Subject: [Hibernate] DistributedCacheConcurrencyStrategy implemented !


> Hi all, got a little time today to investigate Gavin proposition about
> the DistributedCacheConcurrencyStrategy.
> Here is a first shoot, it's based on javagroups to manage a distributed
> lock server ( block provided by javagroup ;) )
>
> Since the lock server is distributed, even if one or more server in the
> cluster fail, the rest can still use the cache.
>
> I used the latest javagroups version available with log4j
> (JavaGroups-2.0.3.log4j.bin.zip).
> Few little things need to be done like using the mutators to set the
> multicast adr / port.
>
> Give it a try , it's fun ;)
>
> Chris
>


----------------------------------------------------------------------------
----


> package cirrus.hibernate.cache;
>
> import org.apache.commons.logging.Log;
> import org.apache.commons.logging.LogFactory;
> import org.javagroups.blocks.LockNotGrantedException;
> import org.javagroups.blocks.LockNotReleasedException;
> import org.javagroups.ChannelException;
>
>
> public class DistributedCacheConcurrencyStrategy implements
CacheConcurrencyStrategy
> {
>     private final CacheConcurrencyStrategy cacheConcurrencyStrategy;
>     private static final Log log =
LogFactory.getLog(DistributedCacheConcurrencyStrategy.class);
>
>     public DistributedCacheConcurrencyStrategy(CacheConcurrencyStrategy
cacheConcurrencyStrategy)
>     {
>         this.cacheConcurrencyStrategy = cacheConcurrencyStrategy;
>     }
>
>
>     public Object get(Object key, long txTimestamp) throws CacheException
>     {
>         Object cachedItem = null;
>         try
>         {
>             DistributedLockServer.getInstance().lock(key);
>             cachedItem = cacheConcurrencyStrategy.get(key,txTimestamp);
>             DistributedLockServer.getInstance().unlock(key);
>         }
>         catch (LockNotGrantedException e)
>         {
>             // The object is already locked
>             if(log.isDebugEnabled()) log.debug("Could not acquire lock on
object: "+key);
>         }
>         catch (ChannelException e)
>         {
>             log.error(e);
>         }
>         catch (LockNotReleasedException e)
>         {
>             log.warn(e);
>         }
>         return cachedItem;
>     }
>
>     public boolean put(Object key, Object value, long txTimestamp) throws
CacheException
>     {
>         boolean isItemCached = false;;
>         try
>         {
>             DistributedLockServer.getInstance().lock(key);
>             isItemCached =
cacheConcurrencyStrategy.put(key,value,txTimestamp);
>             DistributedLockServer.getInstance().unlock(key);
>         }
>         catch (LockNotGrantedException e)
>         {
>             // The object is already locked
>             if(log.isDebugEnabled()) log.debug("Could not acquire lock on
object: "+key);
>         }
>         catch (ChannelException e)
>         {
>             log.error(e);
>         }
>         catch (LockNotReleasedException e)
>         {
>             log.warn(e);
>         }
>         return isItemCached;
>     }
>
>     public void lock(Object key) throws CacheException
>     {
>         try
>         {
>             DistributedLockServer.getInstance().lock(key);
>             cacheConcurrencyStrategy.lock(key);
>             DistributedLockServer.getInstance().unlock(key);
>         }
>         catch (LockNotGrantedException e)
>         {
>             // The object is already locked
>             if(log.isDebugEnabled()) log.debug("Could not acquire lock on
object: "+key);
>         }
>         catch (ChannelException e)
>         {
>             log.error(e);
>         }
>         catch (LockNotReleasedException e)
>         {
>             log.warn(e);
>         }
>     }
>
>     public void release(Object key) throws CacheException
>     {
>         try
>         {
>             DistributedLockServer.getInstance().lock(key);
>             cacheConcurrencyStrategy.release(key);
>             DistributedLockServer.getInstance().unlock(key);
>         }
>         catch (LockNotGrantedException e)
>         {
>             // The object is already locked
>             if(log.isDebugEnabled()) log.debug("Could not acquire lock on
object: "+key);
>         }
>         catch (ChannelException e)
>         {
>             log.error(e);
>         }
>         catch (LockNotReleasedException e)
>         {
>             log.warn(e);
>         }
>     }
> }
>


----------------------------------------------------------------------------
----


> package cirrus.hibernate.cache;
>
> import org.javagroups.JChannel;
> import org.javagroups.ChannelException;
> import org.javagroups.ChannelClosedException;
> import org.javagroups.blocks.*;
> import org.javagroups.blocks.DistributedLockManager;
>
> public class DistributedLockServer
> {
>     private static DistributedLockServer singleton = new
DistributedLockServer();
>
>     private String multiCastAdr = "228.3.11.76";
>     private String multiCastPort = "12345";
>
>     private String SERVER_PROTOCOL_STACK = ""
>         +
"UDP(mcast_addr="+multiCastAdr+";mcast_port="+multiCastPort+";ip_ttl=32;"
>         + "mcast_send_buf_size=150000;mcast_recv_buf_size=80000)"
>         + ":PING(timeout=500;num_initial_members=1)"
>         + ":FD"
>         + ":VERIFY_SUSPECT(timeout=1500)"
>         + ":pbcast.STABLE(desired_avg_gossip=200)"
>         +
":pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800)"
>         + ":FRAG(frag_size=4096)"
>         + ":UNICAST(timeout=5000)"
>         + ":pbcast.GMS(join_timeout=5000;join_retry_timeout=1000;"
>         +     "shun=false;print_local_addr=false)";
>
>     private JChannel channel;
>     private VotingAdapter adapter;
>     private LockManager lockManager;
>     private int lockTimeout;
>
>     // Compute unique identity
>     private String identity = ""+System.currentTimeMillis();
>
>     private DistributedLockServer()
>     {
>         try
>         {
>             channel = new JChannel(SERVER_PROTOCOL_STACK);
>             adapter = new VotingAdapter(channel);
>             channel.connect("voting");
>         }
>         catch (ChannelException e)
>         {
>         }
>         catch (ChannelClosedException e)
>         {
>         }
>
>         lockManager = new
org.javagroups.blocks.DistributedLockManager(adapter, "1");
>         lockTimeout = 5000;
>     }
>
>     public static DistributedLockServer getInstance()
>     {
>         return singleton;
>     }
>
>     public void setMultiCastAdr(String multiCastAdr)
>     {
>         this.multiCastAdr = multiCastAdr;
>     }
>
>     public void setMultiCastPort(String multiCastPort)
>     {
>         this.multiCastPort = multiCastPort;
>     }
>
>     public void setLockTimeout(int lockTimeout)
>     {
>         this.lockTimeout = lockTimeout;
>     }
>
>
>     public void lock(Object key) throws LockNotGrantedException,
ChannelException
>     {
>         lockManager.lock(key,identity,lockTimeout);
>     }
>
>     public void unlock(Object key) throws LockNotReleasedException,
ChannelException
>     {
>         lockManager.unlock(key,identity);
>     }
> }
>