Print

[jandam]

Hello lolocohen,

  again great work on 1.8 release.

I want to create demo application for my boss with JPPF. We have some computation intensive tasks that can be divided to multiple threads or computers. I can't distribute some data access/methods to "node" computer. In multithreaded environment is it no problem because everything runs on one computer.

Best solution for me is new DataProvider. (ClientDataProvider)

Calling on "node":

Object getValue(Object key) throws Exception
- should request value from "client"  on every call (even for same key for simplicity)
- on "client" there will be method
     "Object JPPFTask::provideClientValue(Object key)" - not good from architecture point of view - adding special method to specific DataProvider
     or
  Object ClientDataProvider::provideClientValue(JPPFTask task, Object key)  - this should be accessible/callable only on client machine
     or
  interface ClientDataProviderValueSource {
     Object provideClientValue(JPPFTask task, ClientDataProvider provider, Object key)
}
- compute getValue only once on "client" there can be wrapper to data provider "CacheDataProvider"

void setValue(Object key, Object value) throws Exception
- should throw UnsupportedOperationException - not necessary

------------
public class CacheDataProvider implements DataProvider {
  private final static Object  NULL_VALUE = new Object();

  private final DataProvider  dataProvider;
  private final Map<Object, Object> valueMap = new HashMap<Object, Object>();

  public CacheDataProvider(final DataProvider dataProvider) {
     this.dataProvider = dataProvider;
  }

  public Object getValue(Object key) throws Exception {
    Object result;
    synchronized(valueMap) {
        result = valueMap.get(key);
        if(result == null) {
            result = dataProvider.getValue(key);
            if(result == null) result = NULL_VALUE;
            valueMap.put(key, result);
        }
        if(result == NULL_VALUE)
           return null;
        else
           return result;
    }
  }

  public void setValue(Object key, Object value) throws Exception {
       synchronized(valueMap) {
        if(value == null)
          valueMap.put(key, NULL_VALUE);
       else
          valueMap.put(key, value);
      }
       dataProvider.setValue(key, value);
  }
}

I'm working on other project so I don't have enough time to dig deep into JPPF. I think  that the communication mechanism is already implemented in JPPF.
eg ClassLoader - that according to URL downloads class byte codes from "client". There will be nice to reuse already used ports for communication to implement ClientDataProvider communication.
ClientDataProvider shouldn't distribute same data to all nodes. Every node generate different "key" - (eg 2D coordinates).
"provideClientValue" should be called from multiple threads (ThreadPoolExecutor) because of multiple "node"s that send request for values.

Reason for my RFE:
  1) precompute some data (key for getValue) ~70%
  2) ClientDataProvider::getValue(key) - can be computed only on client  ~10%
  3) final phase ~20% and generate result

"final phase - 3" needs data from 1) and 2). Distributing only phase 1) isn't good idea because it produces huge amout of data (~10-20x more than size of result)

final result is from 10MB - ~128MB of data (uncompressed). I know that there is limitation (according to my tests) of JPPF to 50-100kB JPPFTask result but it isn't problem for me because I can divide data into smaller subtasks

Thank you very much. This is only RFE
  I would like to answer any of your questions.

   Martin

[lolo]

Hello Martin,

I like your idea, and have registered a feature request for it: 2633080 - New data provider to share data stored in the client

After looking at your design, I believe it may be a tad difficult to implement as is.
Instead, I would like to propose an implementation such as this class I wrote today to illustrate the concept: ClientDataProvider.java

The idea is that the data stored in the data provider is transient, so never sent to the node, and that it can be retrieved from the client by using, as you mentioned, an existing capability of the JPPFClassLoader to fetch resources on the client side (see JPPFClassLoader.getResourceAsStream( ) )
I can, relatively easily, add a hook in the class loader mechanism so that, for instance,  resource names that start with [cdp:<data provider uuid>] are interpreted as a key in the client-side data provider with the specified uuid.

What do you think? would this fit your requirements, or is there something that I missed?

Thanks for your time,
-Laurent

[jandam]

Hello lolocohen,

thank you for very fast reply and proposed solution. It is elegant and simple. But it doesn't fit my needs.

Example that better describes my problem:
The key is array of 2D coordinates [double x, double y] and "node" is asking "client" for array of [double z] values.

'Z' values on "client" need to be computed according to 2D coordinates and other parameters from JPPFTask DataProvider.

Code: [Select]

MemoryMapDataProvider providerTaskData = new MemoryMapDataProvider();
taskDataProvider.setValue("scale", 2.5);
taskDataProvider.setValue("transform", AffineTransform.createTranslate(12.0, 13.0));
ClientDataProvider providerTaskZ = new ClientDataProvider();
taskDataProvider.setValue("z-provider", providerTaskZ);                                                // send client data provider to node

providerTaskZ.setValueSource(new ClientValueSource() {

  // taskDataProvider isn't necessary as method parameter
  // scale can be obtained from another source (eg in constructor of ClientValueSource, or from 'key' object)

  public Object getClientValueForKey(final DataProvider taskDataProvider, final Object key) throws Exception {  
    double scale = ((Number)taskDataProvider.getValue("scale")).doubleValue();
    if(key instanceof double[][]) {
      double[][] coordsXY = (double[][])key;
      double[] result = new double[coordsXY.length];
      for(int index = 0; index < coordsXY.length; index++) {
         double x = coordsXY[index][0];
         double y = coordsXY[index][1];
         result[index] = (x + 100.0 * y) * scale;                              // some computation of result depending on key (X, Y) and 'scale' from task
      }
      return result;
   } else
      throw new IllegalArgumentException("Expected coordinate array");
  }
}

JPPFTask task = new JPPFTask(..) {
 ...
  public void run() {
      DataProvider dataProvider = getDataProvider();

      // get task arguments
      double scale = (Double)dataProvider.getValue("scale");
      AffineTransform transform = (AffineTransform)dataProvider.getValue("transform");

     ClientDataProvider zProvider = (ClientDataProvider)dataProvider.getValue("z-provider");

     // lets do something
     double[][] coords = new double[10][2];
     
      coords = [i]function_foo[/i](transform, scale);

     double[] z;
     Object value = zProvider.getValue(coords);     // can return 'value' or Exception for 'key'  when ocured on "client"

     if(value instanceof double[])
         double[] z = (double[])value;
     else if(value instanceof Exception)  {        
         setException((Exception)value));  // handle exception "(Exception)value"
         return; // finish task
     } else {
         setException(new IllegalArgumentException("Wrong data value '" + value + "' from client data provider '" + zProvider + '\''));
         return; // finish task
     }

      result = [i]function_bar[/i](transform, scale, coords, z);
      setResult(result);
  }
}

----
ClientDataProvider can have additional "getValue" method

Code: [Select]

public <T> T getValue(final Class<T> valueClass, final Object key) throws Exception {
  Object value = getValue();

  if(value == null)
    return null;
  else if(valueClass.isAssignableFrom(value.getClass()))
    return valueClass.cast(value);
  else if(value instanceof Exception)
    throw (Exception)value;
  else
    throw new IllegalArgumentException("Wrong data value '" + value + "' expected type '" + valueClass.getName + '\'');
}


your concept of ClientDataProvider partially allows what I need.
  I can serialize my coordinate "key" to Base64 encoded String bud it isn't elegant and there is some overhead (CPU and memory)

key --- serialize --> byte_array -- compress --> compressed_byte_array -- encode Base64 --> base64_string

Base64 takes 1/3 more space than array of bytes

Sorry for any mistakes - code was written directly to editor without any checks just to show what I thinking of and what I need.

I looked at JPPF 1.8 sources.
There are 3 NioServers in "driver" (if I remember ClassNioServer,  JPPFApplicationServer, NodeNioServer) and they are implemented as state machines. And more classes on "node" and "client"
So thats reason why I didn't try implement my version.

Simplest way to extends ClassNioServer and JPPFClassLoader to handle Object as resource name instead of just String. (by changing or extending JPPFResourceWrapper)

Thank you very much

  Martin

[lolo]

Hello Martin,

Sorry for this late reply, I have been working on this feature, and I came up with something that I believe will satisfy your requirements.
I invite you to take a look at the latest version of ClientDataProvider.java.
Basically, it extends MemoryMapDataProvider and adds a method public <V> Object computeValue(Object key, JPPFCallable<V> callable) , where the JPPFCallable object contains a callback method that will be executed on the client.
I have tested it with a simple use case and it is working.

What are your thoughts on this?

Thank you for your time,
-Laurent

[jandam]

Hello,

  thank you very much for work on ClientDataProvider. I'm just finishing one module for our application. So I don't have time with new provider right now.  Maybe on weekend.
I just quickly looked on code repository for your ClientDataProvider and changes to JPPFClassloader in 'node'. (no practical tests).
 Your code is short and elegant and I think that fulfills what I meant. I wish I could have more time for JPPF.
I hope that I will be able to modify our library to make early tests with ClientDataProvider on weekend.

  Once again thank you very much for your time.

   Martin