Add a partitioner component, which combined with a selector component can be used to implement data partitioning, sharding, or data access load balancing for example.

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1231822 13f79535-47bb-0310-9956-ffa450edef68

9 files changed, 319 insertions, 5 deletions

diff --git a/sca-cpp/trunk/components/cache/Makefile.am b/sca-cpp/trunk/components/cache/Makefile.am
index d14bd3a542..1f24d59c72 100644
--- a/sca-cpp/trunk/components/cache/Makefile.am
+++ b/sca-cpp/trunk/components/cache/Makefile.am
@@ -25,10 +25,10 @@ comp_DATA = memcached.prefix
 memcached.prefix: $(top_builddir)/config.status
 	echo ${MEMCACHED_PREFIX} >memcached.prefix
 
-EXTRA_DIST = cache.composite memcache.componentType datacache.componentType memocache.componentType *.scm
+EXTRA_DIST = cache.composite memcache.componentType datacache.componentType memocache.componentType partitioner.componentType *.scm
 
-comp_LTLIBRARIES = libmemcache.la libdatacache.la libmemocache.la
-noinst_DATA = libmemcache${libsuffix} libdatacache${libsuffix} libmemocache${libsuffix}
+comp_LTLIBRARIES = libmemcache.la libdatacache.la libmemocache.la libpartitioner.la
+noinst_DATA = libmemcache${libsuffix} libdatacache${libsuffix} libmemocache${libsuffix} libpartitioner${libsuffix}
 
 libmemcache_la_SOURCES = memcache.cpp
 libmemcache${libsuffix}:
@@ -42,6 +42,10 @@ libmemocache_la_SOURCES = memocache.cpp
 libmemocache${libsuffix}:
 	ln -s .libs/libmemocache${libsuffix}
 
+libpartitioner_la_SOURCES = partitioner.cpp
+libpartitioner${libsuffix}:
+	ln -s .libs/libpartitioner${libsuffix}
+
 memcache_test_SOURCES = memcache-test.cpp
 memcache_test_LDFLAGS = -lxml2
 
diff --git a/sca-cpp/trunk/components/cache/cache.composite b/sca-cpp/trunk/components/cache/cache.composite
index 8a96b6161b..838cbce77f 100644
--- a/sca-cpp/trunk/components/cache/cache.composite
+++ b/sca-cpp/trunk/components/cache/cache.composite
@@ -64,4 +64,35 @@
         </service>
     </component>     
 
+    <component name="partitioner">
+        <implementation.cpp path="." library="libpartitioner"/>
+        <service name="partitioner">
+            <binding.http uri="partitioner"/>
+        </service>
+        <reference name="selector" target="selector"/>
+        <reference name="partition" target="partition1"/>
+        <reference name="partition" target="partition2"/>
+    </component>
+
+    <component name="selector">
+        <implementation.scheme script="select-test.scm"/>
+        <service name="selector">
+            <binding.http uri="selector"/>
+        </service>
+    </component>     
+
+    <component name="partition1">
+        <implementation.scheme script="partition1-test.scm"/>
+        <service name="partition1">
+            <binding.http uri="partition1"/>
+        </service>
+    </component>     
+
+    <component name="partition2">
+        <implementation.scheme script="partition2-test.scm"/>
+        <service name="partition2">
+            <binding.http uri="partition2"/>
+        </service>
+    </component>     
+
 </composite>
diff --git a/sca-cpp/trunk/components/cache/client-test.cpp b/sca-cpp/trunk/components/cache/client-test.cpp
index c44060b7a4..b652ad3a88 100644
--- a/sca-cpp/trunk/components/cache/client-test.cpp
+++ b/sca-cpp/trunk/components/cache/client-test.cpp
@@ -39,6 +39,8 @@ namespace cache {
 const string memcacheuri("http://localhost:8090/memcache");
 const string datacacheuri("http://localhost:8090/datacache");
 const string memocacheuri("http://localhost:8090/memocache");
+const string partition1uri("http://localhost:8090/partitioner/a");
+const string partition2uri("http://localhost:8090/partitioner/b");
 
 bool testCache(const string& uri) {
     http::CURLSession cs("", "", "", "");
@@ -114,6 +116,20 @@ bool testMemocache() {
     return true;
 }
 
+bool testPartitioner() {
+    http::CURLSession cs("", "", "", "");
+
+    const failable<value> res1 = http::get(partition1uri, cs);
+    assert(hasContent(res1));
+    assert(cadr<value>(content(res1)) == string("1"));
+
+    const failable<value> res2 = http::get(partition2uri, cs);
+    assert(hasContent(res2));
+    assert(cadr<value>(content(res2)) == string("2"));
+
+    return true;
+}
+
 struct getLoop {
     const string path;
     const value entry;
@@ -157,6 +173,7 @@ int main() {
     tuscany::cache::testMemcache();
     tuscany::cache::testDatacache();
     tuscany::cache::testMemocache();
+    tuscany::cache::testPartitioner();
     tuscany::cache::testGetPerf();
 
     tuscany::cout << "OK" << tuscany::endl;
diff --git a/sca-cpp/trunk/components/cache/partition1-test.scm b/sca-cpp/trunk/components/cache/partition1-test.scm
new file mode 100644
index 0000000000..547539e2a1
--- /dev/null
+++ b/sca-cpp/trunk/components/cache/partition1-test.scm
@@ -0,0 +1,21 @@
+;  Licensed to the Apache Software Foundation (ASF) under one
+;  or more contributor license agreements.  See the NOTICE file
+;  distributed with this work for additional information
+;  regarding copyright ownership.  The ASF licenses this file
+;  to you under the Apache License, Version 2.0 (the
+;  "License"); you may not use this file except in compliance
+;  with the License.  You may obtain a copy of the License at
+;  
+;    http://www.apache.org/licenses/LICENSE-2.0
+;    
+;  Unless required by applicable law or agreed to in writing,
+;  software distributed under the License is distributed on an
+;  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+;  KIND, either express or implied.  See the License for the
+;  specific language governing permissions and limitations
+;  under the License.
+
+; Partition test case
+
+(define (get key) (list key "1"))
+
diff --git a/sca-cpp/trunk/components/cache/partition2-test.scm b/sca-cpp/trunk/components/cache/partition2-test.scm
new file mode 100644
index 0000000000..60644128df
--- /dev/null
+++ b/sca-cpp/trunk/components/cache/partition2-test.scm
@@ -0,0 +1,21 @@
+;  Licensed to the Apache Software Foundation (ASF) under one
+;  or more contributor license agreements.  See the NOTICE file
+;  distributed with this work for additional information
+;  regarding copyright ownership.  The ASF licenses this file
+;  to you under the Apache License, Version 2.0 (the
+;  "License"); you may not use this file except in compliance
+;  with the License.  You may obtain a copy of the License at
+;  
+;    http://www.apache.org/licenses/LICENSE-2.0
+;    
+;  Unless required by applicable law or agreed to in writing,
+;  software distributed under the License is distributed on an
+;  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+;  KIND, either express or implied.  See the License for the
+;  specific language governing permissions and limitations
+;  under the License.
+
+; Partition test case
+
+(define (get key) (list key "2"))
+
diff --git a/sca-cpp/trunk/components/cache/partitioner.componentType b/sca-cpp/trunk/components/cache/partitioner.componentType
new file mode 100644
index 0000000000..9ff2d4f0b5
--- /dev/null
+++ b/sca-cpp/trunk/components/cache/partitioner.componentType
@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+-->
+<componentType xmlns="http://docs.oasis-open.org/ns/opencsa/sca/200912"
+  xmlns:t="http://tuscany.apache.org/xmlns/sca/1.1"
+  targetNamespace="http://tuscany.apache.org/xmlns/sca/components">
+        
+    <service name="partitioner"/>
+    <reference name="selector"/>
+    <reference name="partition" multiplicity="0..n"/>
+
+</composite>
diff --git a/sca-cpp/trunk/components/cache/partitioner.cpp b/sca-cpp/trunk/components/cache/partitioner.cpp
new file mode 100644
index 0000000000..609d42f494
--- /dev/null
+++ b/sca-cpp/trunk/components/cache/partitioner.cpp
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/* $Rev$ $Date$ */
+
+/**
+ * A partitioner component implementation which forwards data access requests to a
+ * dynamically selected data store component. The selection is externalized, performed
+ * by a selector component, responsible for selecting the target data store given the
+ * data access request key and a list of references to available data store components.
+ * This pattern can be used for sharding or load balancing for example.
+ */
+
+#define WANT_HTTPD_LOG 1
+#include "string.hpp"
+#include "function.hpp"
+#include "list.hpp"
+#include "value.hpp"
+#include "monad.hpp"
+
+namespace tuscany {
+namespace partitioner {
+
+/**
+ * Return the target partition for a key.
+ */
+const failable<lambda<value(const list<value>&)> > partition(const value& key, const lambda<value(const list<value>&)> selector, const list<value>& partitions) {
+
+    // Call the selector component to convert the given key to a partition number
+    const value p = selector(mklist<value>("get", key, partitions));
+    if (isNil(p)) {
+        ostringstream os;
+        os << "Couldn't get partition number: " << key;
+        return mkfailure<lambda<value(const list<value>&)> >(str(os), false);
+    }
+    return (const lambda<value(const list<value>&)>)p;
+}
+
+/**
+ * Get an item from a partition.
+ */
+const failable<value> get(const value& key, const lambda<value(const list<value>&)> selector, const list<value>& partitions) {
+
+    // Select partition
+    const failable<lambda<value(const list<value>&)> > p = partition(key, selector, partitions);
+    if (!hasContent(p))
+        return mkfailure<value>(reason(p));
+
+    // Get from selected partition
+    const value val = content(p)(mklist<value>("get", key));
+    if (isNil(val)) {
+        ostringstream os;
+        os << "Couldn't get entry from partition: " << key;
+        return mkfailure<value>(str(os), false);
+    }
+
+    return val;
+}
+
+/**
+ * Post an item to a partition.
+ */
+const failable<value> post(const value& key, const value& val, const lambda<value(const list<value>&)> selector, const list<value>& partitions) {
+    const value id = append<value>(key, mklist(mkuuid()));
+
+    // Select partition
+    const failable<lambda<value(const list<value>&)> > p = partition(id, selector, partitions);
+    if (!hasContent(p))
+        return mkfailure<value>(reason(p));
+
+    // Put into select partition
+    content(p)(mklist<value>("put", id, val));
+
+    return id;
+}
+
+/**
+ * Put an item into a partition.
+ */
+const failable<value> put(const value& key, const value& val, const lambda<value(const list<value>&)> selector, const list<value>& partitions) {
+
+    // Select partition
+    const failable<lambda<value(const list<value>&)> > p = partition(key, selector, partitions);
+    if (!hasContent(p))
+        return mkfailure<value>(reason(p));
+
+    // Put into selected partition
+    content(p)(mklist<value>("put", key, val));
+
+    return value(true);
+}
+
+/**
+ * Delete an item from a partition.
+ */
+const failable<value> del(const value& key, const lambda<value(const list<value>&)> selector, const list<value>& partitions) {
+
+    // Select partition
+    const failable<lambda<value(const list<value>&)> > p = partition(key, selector, partitions);
+    if (!hasContent(p))
+        return mkfailure<value>(reason(p));
+
+    // Delete from selected partition
+    content(p)(mklist<value>("delete", key));
+
+    return value(true);
+}
+
+}
+}
+
+extern "C" {
+
+const tuscany::value apply(const tuscany::list<tuscany::value>& params) {
+    const tuscany::value func(car(params));
+    if (func == "get")
+        return tuscany::partitioner::get(cadr(params), caddr(params), cdddr(params));
+    if (func == "post")
+        return tuscany::partitioner::post(cadr(params), caddr(params), cadddr(params), cddddr(params));
+    if (func == "put")
+        return tuscany::partitioner::put(cadr(params), caddr(params), cadddr(params), cddddr(params));
+    if (func == "delete")
+        return tuscany::partitioner::del(cadr(params), caddr(params), cdddr(params));
+    return tuscany::mkfailure<tuscany::value>();
+}
+
+}
diff --git a/sca-cpp/trunk/components/cache/select-test.scm b/sca-cpp/trunk/components/cache/select-test.scm
new file mode 100644
index 0000000000..9baa82a5da
--- /dev/null
+++ b/sca-cpp/trunk/components/cache/select-test.scm
@@ -0,0 +1,21 @@
+;  Licensed to the Apache Software Foundation (ASF) under one
+;  or more contributor license agreements.  See the NOTICE file
+;  distributed with this work for additional information
+;  regarding copyright ownership.  The ASF licenses this file
+;  to you under the Apache License, Version 2.0 (the
+;  "License"); you may not use this file except in compliance
+;  with the License.  You may obtain a copy of the License at
+;  
+;    http://www.apache.org/licenses/LICENSE-2.0
+;    
+;  Unless required by applicable law or agreed to in writing,
+;  software distributed under the License is distributed on an
+;  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+;  KIND, either express or implied.  See the License for the
+;  specific language governing permissions and limitations
+;  under the License.
+
+; Partition selector test case
+
+(define (get key partitions) (if (= (car key) "a") (car partitions) (cadr partitions)))
+
diff --git a/sca-cpp/trunk/kernel/list.hpp b/sca-cpp/trunk/kernel/list.hpp
index df7bc27c03..60dcbb083a 100644
--- a/sca-cpp/trunk/kernel/list.hpp
+++ b/sca-cpp/trunk/kernel/list.hpp
@@ -362,20 +362,48 @@ template<typename T> const T cadddddddr(const list<T>& p) {
 }
 
 /**
- * Returns the cdr of a cdr of a list.
+ * Returns a list of elements from the 3rd to the end of a list.
  */
 template<typename T> const list<T> cddr(const list<T>& p) {
     return cdr(cdr(p));
 }
 
 /**
- * Returns the cdr of a cdr of the cdr of a list.
+ * Returns a list of elements from the 4th to the end of a list.
  */
 template<typename T> const list<T> cdddr(const list<T>& p) {
     return cdr(cdr(cdr(p)));
 }
 
 /**
+ * Returns a list of elements from the 5th to the end of a list.
+ */
+template<typename T> const list<T> cddddr(const list<T>& p) {
+    return cdr(cdr(cdr(cdr(p))));
+}
+
+/**
+ * Returns a list of elements from the 6th to the end of a list.
+ */
+template<typename T> const list<T> cdddddr(const list<T>& p) {
+    return cdr(cdr(cdr(cdr(cdr(p)))));
+}
+
+/**
+ * Returns a list of elements from the 7th to the end of a list.
+ */
+template<typename T> const list<T> cddddddr(const list<T>& p) {
+    return cdr(cdr(cdr(cdr(cdr(cdr(p))))));
+}
+
+/**
+ * Returns a list of elements from the 8th to the end of a list.
+ */
+template<typename T> const list<T> cdddddddr(const list<T>& p) {
+    return cdr(cdr(cdr(cdr(cdr(cdr(cdr(p)))))));
+}
+
+/**
  * Returns the length of a list.
  */
 template<typename T> struct lengthRef {