You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by is...@apache.org on 2017/05/29 14:30:47 UTC
[1/2] ignite git commit: IGNITE-3355: Implemented Compute::Call() for
C++
Repository: ignite
Updated Branches:
refs/heads/master 0f8af137c -> f9c96de57
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_holder.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_holder.h b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_holder.h
new file mode 100644
index 0000000..e218e36
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_holder.h
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::impl::compute::ComputeJobHolder class template.
+ */
+
+#ifndef _IGNITE_IMPL_COMPUTE_COMPUTE_JOB_HOLDER
+#define _IGNITE_IMPL_COMPUTE_COMPUTE_JOB_HOLDER
+
+#include <ignite/impl/binary/binary_writer_impl.h>
+#include <ignite/impl/compute/compute_job_result.h>
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ /**
+ * Compute job holder. Internal helper class.
+ * Used to handle jobs in general way, without specific types.
+ */
+ class ComputeJobHolder
+ {
+ public:
+ /**
+ * Destructor.
+ */
+ virtual ~ComputeJobHolder()
+ {
+ // No-op.
+ }
+
+ /**
+ * Execute job locally.
+ */
+ virtual void ExecuteLocal() = 0;
+
+ /**
+ * Execute job remote.
+ *
+ * @param writer Writer.
+ */
+ virtual void ExecuteRemote(binary::BinaryWriterImpl& writer) = 0;
+ };
+
+ /**
+ * Compute job holder. Internal class.
+ *
+ * @tparam F Actual job type.
+ * @tparam R Job return type.
+ */
+ template<typename F, typename R>
+ class ComputeJobHolderImpl : public ComputeJobHolder
+ {
+ public:
+ typedef R ResultType;
+ typedef F JobType;
+
+ /**
+ * Constructor.
+ *
+ * @param job Job.
+ */
+ ComputeJobHolderImpl(JobType job) :
+ job(job)
+ {
+ // No-op.
+ }
+
+ /**
+ * Destructor.
+ */
+ virtual ~ComputeJobHolderImpl()
+ {
+ // No-op.
+ }
+
+ const ComputeJobResult<ResultType>& GetResult()
+ {
+ return res;
+ }
+
+ virtual void ExecuteLocal()
+ {
+ try
+ {
+ res.SetResult(job.Call());
+ }
+ catch (const IgniteError& err)
+ {
+ res.SetError(err);
+ }
+ catch (const std::exception& err)
+ {
+ res.SetError(IgniteError(IgniteError::IGNITE_ERR_STD, err.what()));
+ }
+ catch (...)
+ {
+ res.SetError(IgniteError(IgniteError::IGNITE_ERR_UNKNOWN,
+ "Unknown error occurred during call."));
+ }
+ }
+
+ virtual void ExecuteRemote(binary::BinaryWriterImpl& writer)
+ {
+ ExecuteLocal();
+
+ res.Write(writer);
+ }
+
+ private:
+ /** Result. */
+ ComputeJobResult<ResultType> res;
+
+ /** Job. */
+ JobType job;
+ };
+ }
+ }
+}
+
+#endif //_IGNITE_IMPL_COMPUTE_COMPUTE_JOB_HOLDER
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_result.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_result.h b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_result.h
new file mode 100644
index 0000000..5bcb762
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_job_result.h
@@ -0,0 +1,161 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::impl::compute::ComputeJobResult class template.
+ */
+
+#ifndef _IGNITE_IMPL_COMPUTE_COMPUTE_JOB_RESULT
+#define _IGNITE_IMPL_COMPUTE_COMPUTE_JOB_RESULT
+
+#include <memory>
+#include <sstream>
+
+#include <ignite/common/promise.h>
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ /**
+ * Used to hold compute job result.
+ */
+ template<typename R>
+ class ComputeJobResult
+ {
+ public:
+ typedef R ResultType;
+ /**
+ * Default constructor.
+ */
+ ComputeJobResult() :
+ res(),
+ err()
+ {
+ // No-op.
+ }
+
+ /**
+ * Set result value.
+ *
+ * @param val Value to set as a result.
+ */
+ void SetResult(const ResultType& val)
+ {
+ res = val;
+ }
+
+ /**
+ * Set error.
+ *
+ * @param error Error to set.
+ */
+ void SetError(const IgniteError error)
+ {
+ err = error;
+ }
+
+ /**
+ * Set promise to a state which corresponds to result.
+ *
+ * @param promise Promise, which state to set.
+ */
+ void SetPromise(common::Promise<ResultType>& promise)
+ {
+ if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
+ promise.SetError(err);
+ else
+ promise.SetValue(std::auto_ptr<ResultType>(new ResultType(res)));
+ }
+
+ /**
+ * Write using writer.
+ *
+ * @param writer Writer.
+ */
+ void Write(binary::BinaryWriterImpl& writer)
+ {
+ if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
+ {
+ // Fail
+ writer.WriteBool(false);
+
+ // Native Exception
+ writer.WriteBool(true);
+
+ writer.WriteObject<IgniteError>(err);
+ }
+ else
+ {
+ // Success
+ writer.WriteBool(true);
+
+ writer.WriteObject<ResultType>(res);
+ }
+ }
+
+ /**
+ * Read using reader.
+ *
+ * @param reader Reader.
+ */
+ void Read(binary::BinaryReaderImpl& reader)
+ {
+ bool success = reader.ReadBool();
+
+ if (success)
+ {
+ res = reader.ReadObject<ResultType>();
+
+ err = IgniteError();
+ }
+ else
+ {
+ bool native = reader.ReadBool();
+
+ if (native)
+ err = reader.ReadObject<IgniteError>();
+ else
+ {
+ std::stringstream buf;
+
+ buf << reader.ReadObject<std::string>() << " : ";
+ buf << reader.ReadObject<std::string>() << ", ";
+ buf << reader.ReadObject<std::string>();
+
+ std::string msg = buf.str();
+
+ err = IgniteError(IgniteError::IGNITE_ERR_GENERIC, msg.c_str());
+ }
+ }
+ }
+
+ private:
+ /** Result. */
+ ResultType res;
+
+ /** Erorr. */
+ IgniteError err;
+ };
+ }
+ }
+}
+
+#endif //_IGNITE_IMPL_COMPUTE_COMPUTE_JOB_RESULT
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/compute/compute_task_holder.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/compute/compute_task_holder.h b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_task_holder.h
new file mode 100644
index 0000000..bdd7513
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_task_holder.h
@@ -0,0 +1,213 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::impl::compute::ComputeTaskHolder class and
+ * ignite::impl::compute::ComputeTaskHolderImpl class template.
+ */
+
+#ifndef _IGNITE_IMPL_COMPUTE_COMPUTE_TASK_IMPL
+#define _IGNITE_IMPL_COMPUTE_COMPUTE_TASK_IMPL
+
+#include <stdint.h>
+
+#include <ignite/common/promise.h>
+#include <ignite/impl/compute/compute_job_result.h>
+#include <ignite/impl/compute/compute_job_holder.h>
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ struct ComputeJobResultPolicy
+ {
+ enum Type
+ {
+ /**
+ * Wait for results if any are still expected. If all results have been received -
+ * it will start reducing results.
+ */
+ WAIT = 0,
+
+ /**
+ * Ignore all not yet received results and start reducing results.
+ */
+ REDUCE = 1,
+
+ /**
+ * Fail-over job to execute on another node.
+ */
+ FAILOVER = 2
+ };
+ };
+
+ /**
+ * Compute task holder. Internal helper class.
+ * Used to handle tasks in general way, without specific types.
+ */
+ class ComputeTaskHolder
+ {
+ public:
+ /**
+ * Constructor.
+ *
+ * @param handle Job handle.
+ */
+ ComputeTaskHolder(int64_t handle) :
+ handle(handle)
+ {
+ // No-op.
+ }
+
+ /**
+ * Destructor.
+ */
+ virtual ~ComputeTaskHolder()
+ {
+ // No-op.
+ }
+
+ /**
+ * Process local job result.
+ *
+ * @param job Job.
+ * @return Policy.
+ */
+ virtual int32_t JobResultLocal(ComputeJobHolder& job) = 0;
+
+ /**
+ * Process remote job result.
+ *
+ * @param job Job.
+ * @param reader Reader for stream with result.
+ * @return Policy.
+ */
+ virtual int32_t JobResultRemote(ComputeJobHolder& job, binary::BinaryReaderImpl& reader) = 0;
+
+ /**
+ * Reduce results of related jobs.
+ */
+ virtual void Reduce() = 0;
+
+ /**
+ * Get related job handle.
+ *
+ * @return Job handle.
+ */
+ int64_t GetJobHandle()
+ {
+ return handle;
+ }
+
+ private:
+ /** Related job handle. */
+ int64_t handle;
+ };
+
+ /**
+ * Compute task holder type-specific implementation.
+ */
+ template<typename F, typename R>
+ class ComputeTaskHolderImpl : public ComputeTaskHolder
+ {
+ public:
+ typedef F JobType;
+ typedef R ResultType;
+
+ /**
+ * Constructor.
+ *
+ * @param handle Job handle.
+ */
+ ComputeTaskHolderImpl(int64_t handle) :
+ ComputeTaskHolder(handle)
+ {
+ // No-op.
+ }
+
+ /**
+ * Destructor.
+ */
+ virtual ~ComputeTaskHolderImpl()
+ {
+ // No-op.
+ }
+
+ /**
+ * Process local job result.
+ *
+ * @param job Job.
+ * @return Policy.
+ */
+ virtual int32_t JobResultLocal(ComputeJobHolder& job)
+ {
+ typedef ComputeJobHolderImpl<JobType, ResultType> ActualComputeJobHolder;
+
+ ActualComputeJobHolder& job0 = static_cast<ActualComputeJobHolder&>(job);
+
+ res = job0.GetResult();
+
+ return ComputeJobResultPolicy::WAIT;
+ }
+
+ /**
+ * Process remote job result.
+ *
+ * @param job Job.
+ * @param reader Reader for stream with result.
+ * @return Policy.
+ */
+ virtual int32_t JobResultRemote(ComputeJobHolder& job, binary::BinaryReaderImpl& reader)
+ {
+ res.Read(reader);
+
+ return ComputeJobResultPolicy::WAIT;
+ }
+
+ /**
+ * Reduce results of related jobs.
+ */
+ virtual void Reduce()
+ {
+ res.SetPromise(promise);
+ }
+
+ /**
+ * Get result promise.
+ *
+ * @return Reference to result promise.
+ */
+ common::Promise<ResultType>& GetPromise()
+ {
+ return promise;
+ }
+
+ private:
+ /** Result. */
+ ComputeJobResult<ResultType> res;
+
+ /** Task result promise. */
+ common::Promise<ResultType> promise;
+ };
+ }
+ }
+}
+
+#endif //_IGNITE_IMPL_COMPUTE_COMPUTE_TASK_IMPL
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/ignite_binding_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/ignite_binding_impl.h b/modules/platforms/cpp/core/include/ignite/impl/ignite_binding_impl.h
index a99855a..d0de432 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/ignite_binding_impl.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/ignite_binding_impl.h
@@ -52,6 +52,8 @@ namespace ignite
CACHE_ENTRY_FILTER_CREATE = 2,
CACHE_ENTRY_FILTER_APPLY = 3,
+
+ COMPUTE_JOB_CREATE = 4,
};
};
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/ignite_environment.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/ignite_environment.h b/modules/platforms/cpp/core/include/ignite/impl/ignite_environment.h
index e3cb859..13f7b80 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/ignite_environment.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/ignite_environment.h
@@ -209,6 +209,75 @@ namespace ignite
*/
common::concurrent::SharedPointer<IgniteBindingImpl> GetBinding() const;
+ /**
+ * Get processor compute.
+ *
+ * @param proj Projection.
+ * @return Processor compute.
+ */
+ jobject GetProcessorCompute(jobject proj);
+
+ /**
+ * Locally execute compute job.
+ *
+ * @param jobHandle Job handle.
+ */
+ void ComputeJobExecuteLocal(int64_t jobHandle);
+
+ /**
+ * Locally commit job execution result for the task.
+ *
+ * @param taskHandle Task handle.
+ * @param jobHandle Job handle.
+ * @return Reduce politics.
+ */
+ int32_t ComputeTaskLocalJobResult(int64_t taskHandle, int64_t jobHandle);
+
+ /**
+ * Reduce compute task.
+ *
+ * @param taskHandle Task handle.
+ */
+ void ComputeTaskReduce(int64_t taskHandle);
+
+ /**
+ * Complete compute task.
+ *
+ * @param taskHandle Task handle.
+ */
+ void ComputeTaskComplete(int64_t taskHandle);
+
+ /**
+ * Create compute job.
+ *
+ * @param mem Memory.
+ * @return Job handle.
+ */
+ int64_t ComputeJobCreate(common::concurrent::SharedPointer<interop::InteropMemory>& mem);
+
+ /**
+ * Execute compute job.
+ *
+ * @param mem Memory.
+ * @return Job handle.
+ */
+ void ComputeJobExecute(common::concurrent::SharedPointer<interop::InteropMemory>& mem);
+
+ /**
+ * Destroy compute job.
+ *
+ * @param jobHandle Job handle to destroy.
+ */
+ void ComputeJobDestroy(int64_t jobHandle);
+
+ /**
+ * Consume result of remote job execution.
+ *
+ * @param mem Memory containing result.
+ * @return Reduce policy.
+ */
+ int32_t ComputeTaskJobResult(common::concurrent::SharedPointer<interop::InteropMemory>& mem);
+
private:
/** Node configuration. */
IgniteConfiguration* cfg;
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/ignite_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/ignite_impl.h b/modules/platforms/cpp/core/include/ignite/impl/ignite_impl.h
index 5b1f527..baddec4 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/ignite_impl.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/ignite_impl.h
@@ -22,10 +22,11 @@
#include <ignite/jni/java.h>
#include <ignite/common/utils.h>
+#include <ignite/impl/ignite_environment.h>
#include <ignite/impl/cache/cache_impl.h>
#include <ignite/impl/transactions/transactions_impl.h>
#include <ignite/impl/cluster/cluster_group_impl.h>
-#include <ignite/impl/ignite_environment.h>
+#include <ignite/impl/compute/compute_impl.h>
namespace ignite
{
@@ -38,7 +39,8 @@ namespace ignite
{
typedef common::concurrent::SharedPointer<IgniteEnvironment> SP_IgniteEnvironment;
typedef common::concurrent::SharedPointer<transactions::TransactionsImpl> SP_TransactionsImpl;
- typedef common::concurrent::SharedPointer<cluster::ClusterGroupImpl> SP_ClusterGroupImpl;
+ typedef common::concurrent::SharedPointer<compute::ComputeImpl> SP_ComputeImpl;
+ typedef common::concurrent::SharedPointer<IgniteBindingImpl> SP_IgniteBindingImpl;
public:
/**
* Constructor used to create new instance.
@@ -154,7 +156,7 @@ namespace ignite
*
* @return IgniteBinding class instance.
*/
- common::concurrent::SharedPointer<IgniteBindingImpl> GetBinding();
+ SP_IgniteBindingImpl GetBinding();
/**
* Get instance of the implementation from the proxy class.
@@ -185,7 +187,7 @@ namespace ignite
*
* @return TransactionsImpl instance.
*/
- SP_TransactionsImpl GetTransactions() const
+ SP_TransactionsImpl GetTransactions()
{
return txImpl;
}
@@ -195,11 +197,18 @@ namespace ignite
*
* @return ClusterGroupImpl instance.
*/
- SP_ClusterGroupImpl GetProjection() const
+ cluster::SP_ClusterGroupImpl GetProjection()
{
return prjImpl;
}
+ /**
+ * Get compute.
+ *
+ * @return ComputeImpl instance.
+ */
+ SP_ComputeImpl GetCompute();
+
private:
/**
* Get transactions internal call.
@@ -213,7 +222,7 @@ namespace ignite
*
* @return ClusterGroupImpl instance.
*/
- SP_ClusterGroupImpl InternalGetProjection(IgniteError &err);
+ cluster::SP_ClusterGroupImpl InternalGetProjection(IgniteError &err);
/** Environment. */
SP_IgniteEnvironment env;
@@ -225,7 +234,7 @@ namespace ignite
SP_TransactionsImpl txImpl;
/** Projection implementation. */
- SP_ClusterGroupImpl prjImpl;
+ cluster::SP_ClusterGroupImpl prjImpl;
IGNITE_NO_COPY_ASSIGNMENT(IgniteImpl)
};
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/interop/interop_target.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/interop/interop_target.h b/modules/platforms/cpp/core/include/ignite/impl/interop/interop_target.h
index f9b2b7f..0384dcc 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/interop/interop_target.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/interop/interop_target.h
@@ -62,7 +62,7 @@ namespace ignite
/**
* Destructor.
*/
- ~InteropTarget();
+ virtual ~InteropTarget();
/**
* Internal out operation.
@@ -135,6 +135,15 @@ namespace ignite
OperationResult::Type InStreamOutLong(int32_t opType, InteropMemory& outInMem, IgniteError& err);
/**
+ * In stream out object operation.
+ *
+ * @param opType Type of operation.
+ * @param outInMem Input and output memory.
+ * @return Java object references.
+ */
+ jobject InStreamOutObject(int32_t opType, InteropMemory& outInMem);
+
+ /**
* Internal out-in operation.
*
* @param opType Operation type.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/project/vs/core.vcxproj
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/project/vs/core.vcxproj b/modules/platforms/cpp/core/project/vs/core.vcxproj
index 5cd49f3..9911ffe 100644
--- a/modules/platforms/cpp/core/project/vs/core.vcxproj
+++ b/modules/platforms/cpp/core/project/vs/core.vcxproj
@@ -208,6 +208,8 @@
<ClInclude Include="..\..\include\ignite\cache\query\query_sql.h" />
<ClInclude Include="..\..\include\ignite\cache\query\query_sql_fields.h" />
<ClInclude Include="..\..\include\ignite\cache\query\query_text.h" />
+ <ClInclude Include="..\..\include\ignite\compute\compute.h" />
+ <ClInclude Include="..\..\include\ignite\compute\compute_func.h" />
<ClInclude Include="..\..\include\ignite\ignite_binding_context.h" />
<ClInclude Include="..\..\include\ignite\ignite.h" />
<ClInclude Include="..\..\include\ignite\ignite_configuration.h" />
@@ -225,6 +227,11 @@
<ClInclude Include="..\..\include\ignite\impl\cache\query\query_fields_row_impl.h" />
<ClInclude Include="..\..\include\ignite\impl\cache\query\query_impl.h" />
<ClInclude Include="..\..\include\ignite\impl\cluster\cluster_group_impl.h" />
+ <ClInclude Include="..\..\include\ignite\impl\compute\cancelable_impl.h" />
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_impl.h" />
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_job_holder.h" />
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_job_result.h" />
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_task_holder.h" />
<ClInclude Include="..\..\include\ignite\impl\helpers.h" />
<ClInclude Include="..\..\include\ignite\impl\ignite_environment.h" />
<ClInclude Include="..\..\include\ignite\impl\ignite_impl.h" />
@@ -251,6 +258,8 @@
<ClCompile Include="..\..\src\impl\cache\query\continuous\continuous_query_handle_impl.cpp" />
<ClCompile Include="..\..\src\impl\cache\query\query_impl.cpp" />
<ClCompile Include="..\..\src\impl\cluster\cluster_group_impl.cpp" />
+ <ClCompile Include="..\..\src\impl\compute\cancelable_impl.cpp" />
+ <ClCompile Include="..\..\src\impl\compute\compute_impl.cpp" />
<ClCompile Include="..\..\src\impl\ignite_binding_impl.cpp" />
<ClCompile Include="..\..\src\impl\ignite_environment.cpp" />
<ClCompile Include="..\..\src\impl\ignite_impl.cpp" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/project/vs/core.vcxproj.filters
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/project/vs/core.vcxproj.filters b/modules/platforms/cpp/core/project/vs/core.vcxproj.filters
index 98099a9..7b84494 100644
--- a/modules/platforms/cpp/core/project/vs/core.vcxproj.filters
+++ b/modules/platforms/cpp/core/project/vs/core.vcxproj.filters
@@ -55,6 +55,12 @@
<ClCompile Include="..\..\src\impl\ignite_binding_impl.cpp">
<Filter>Code\impl</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\impl\compute\compute_impl.cpp">
+ <Filter>Code\impl\compute</Filter>
+ </ClCompile>
+ <ClCompile Include="..\..\src\impl\compute\cancelable_impl.cpp">
+ <Filter>Code\impl\compute</Filter>
+ </ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\include\ignite\impl\cache\cache_impl.h">
@@ -210,6 +216,27 @@
<ClInclude Include="..\..\include\ignite\impl\helpers.h">
<Filter>Code\impl</Filter>
</ClInclude>
+ <ClInclude Include="..\..\include\ignite\compute\compute.h">
+ <Filter>Code\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_impl.h">
+ <Filter>Code\impl\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\compute\compute_func.h">
+ <Filter>Code\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\compute\cancelable_impl.h">
+ <Filter>Code\impl\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_job_holder.h">
+ <Filter>Code\impl\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_job_result.h">
+ <Filter>Code\impl\compute</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\compute\compute_task_holder.h">
+ <Filter>Code\impl\compute</Filter>
+ </ClInclude>
</ItemGroup>
<ItemGroup>
<Filter Include="Code">
@@ -257,5 +284,11 @@
<Filter Include="Code\impl\cache\event">
<UniqueIdentifier>{9c5e9732-755a-4553-8926-b4cf3b6abaf3}</UniqueIdentifier>
</Filter>
+ <Filter Include="Code\compute">
+ <UniqueIdentifier>{f1b7ced1-0e6e-4e07-a5b6-04b076797c6f}</UniqueIdentifier>
+ </Filter>
+ <Filter Include="Code\impl\compute">
+ <UniqueIdentifier>{ef20cfe1-cd30-429d-a241-575696df8399}</UniqueIdentifier>
+ </Filter>
</ItemGroup>
</Project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/ignite.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/ignite.cpp b/modules/platforms/cpp/core/src/ignite.cpp
index 2665916..9c42f1d 100644
--- a/modules/platforms/cpp/core/src/ignite.cpp
+++ b/modules/platforms/cpp/core/src/ignite.cpp
@@ -55,6 +55,11 @@ namespace ignite
return transactions::Transactions(txImpl);
}
+ compute::Compute Ignite::GetCompute()
+ {
+ return compute::Compute(impl.Get()->GetCompute());
+ }
+
IgniteBinding Ignite::GetBinding()
{
return impl.Get()->GetBinding();
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/cluster/cluster_group_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/cluster/cluster_group_impl.cpp b/modules/platforms/cpp/core/src/impl/cluster/cluster_group_impl.cpp
index 1bddeac..c34e828 100644
--- a/modules/platforms/cpp/core/src/impl/cluster/cluster_group_impl.cpp
+++ b/modules/platforms/cpp/core/src/impl/cluster/cluster_group_impl.cpp
@@ -37,7 +37,7 @@ namespace ignite
ClusterGroupImpl::ClusterGroupImpl(SP_IgniteEnvironment env, jobject javaRef) :
InteropTarget(env, javaRef)
{
- // No-op.
+ computeImpl = InternalGetCompute();
}
ClusterGroupImpl::~ClusterGroupImpl()
@@ -45,22 +45,33 @@ namespace ignite
// No-op.
}
- ClusterGroupImpl::SP_ClusterGroupImpl ClusterGroupImpl::ForServers(IgniteError& err)
+ SP_ClusterGroupImpl ClusterGroupImpl::ForServers()
{
- JniErrorInfo jniErr;
+ IgniteError err;
jobject res = InOpObject(Command::FOR_SERVERS, err);
- if (jniErr.code != java::IGNITE_JNI_ERR_SUCCESS)
- return SP_ClusterGroupImpl();
+ IgniteError::ThrowIfNeeded(err);
return FromTarget(res);
}
- ClusterGroupImpl::SP_ClusterGroupImpl ClusterGroupImpl::FromTarget(jobject javaRef)
+ ClusterGroupImpl::SP_ComputeImpl ClusterGroupImpl::GetCompute()
+ {
+ return computeImpl;
+ }
+
+ SP_ClusterGroupImpl ClusterGroupImpl::FromTarget(jobject javaRef)
{
return SP_ClusterGroupImpl(new ClusterGroupImpl(GetEnvironmentPointer(), javaRef));
}
+
+ ClusterGroupImpl::SP_ComputeImpl ClusterGroupImpl::InternalGetCompute()
+ {
+ jobject computeProc = GetEnvironment().GetProcessorCompute(GetTarget());
+
+ return SP_ComputeImpl(new compute::ComputeImpl(GetEnvironmentPointer(), computeProc));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/compute/cancelable_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/compute/cancelable_impl.cpp b/modules/platforms/cpp/core/src/impl/compute/cancelable_impl.cpp
new file mode 100644
index 0000000..6e61cc8
--- /dev/null
+++ b/modules/platforms/cpp/core/src/impl/compute/cancelable_impl.cpp
@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+#include <ignite/impl/compute/cancelable_impl.h>
+
+using namespace ignite::common::concurrent;
+
+namespace
+{
+ /**
+ * Operation type.
+ */
+ struct Operation
+ {
+ enum Type
+ {
+ Cancel = 1
+ };
+ };
+}
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ CancelableImpl::CancelableImpl(SharedPointer<IgniteEnvironment> env, jobject javaRef) :
+ InteropTarget(env, javaRef),
+ Cancelable()
+ {
+ // No-op.
+ }
+
+ void CancelableImpl::Cancel()
+ {
+ IgniteError err;
+
+ OutInOpLong(Operation::Cancel, 0, err);
+
+ IgniteError::ThrowIfNeeded(err);
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/compute/compute_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/compute/compute_impl.cpp b/modules/platforms/cpp/core/src/impl/compute/compute_impl.cpp
new file mode 100644
index 0000000..591dd1f
--- /dev/null
+++ b/modules/platforms/cpp/core/src/impl/compute/compute_impl.cpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+#include <ignite/impl/compute/compute_impl.h>
+
+using namespace ignite::common::concurrent;
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ ComputeImpl::ComputeImpl(SharedPointer<IgniteEnvironment> env, jobject javaRef) :
+ InteropTarget(env, javaRef)
+ {
+ // No-op.
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/ignite_environment.cpp b/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
index 2231003..4e78f09 100644
--- a/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
+++ b/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
@@ -20,10 +20,10 @@
#include <ignite/impl/binary/binary_type_updater_impl.h>
#include <ignite/impl/module_manager.h>
#include <ignite/impl/ignite_binding_impl.h>
+#include <ignite/impl/compute/compute_task_holder.h>
#include <ignite/binary/binary.h>
#include <ignite/cache/query/continuous/continuous_query.h>
-#include <ignite/ignite_binding.h>
#include <ignite/ignite_binding_context.h>
#include <ignite/impl/ignite_environment.h>
@@ -47,13 +47,21 @@ namespace ignite
enum Type
{
CACHE_INVOKE = 8,
+ COMPUTE_TASK_JOB_RESULT = 10,
+ COMPUTE_TASK_REDUCE = 11,
+ COMPUTE_TASK_COMPLETE = 12,
+ COMPUTE_JOB_CREATE = 14,
+ COMPUTE_JOB_EXECUTE = 15,
+ COMPUTE_JOB_DESTROY = 17,
CONTINUOUS_QUERY_LISTENER_APPLY = 18,
CONTINUOUS_QUERY_FILTER_CREATE = 19,
CONTINUOUS_QUERY_FILTER_APPLY = 20,
CONTINUOUS_QUERY_FILTER_RELEASE = 21,
REALLOC = 36,
ON_START = 49,
- ON_STOP = 50
+ ON_STOP = 50,
+ COMPUTE_TASK_LOCAL_JOB_RESULT = 60,
+ COMPUTE_JOB_EXECUTE_LOCAL = 61
};
};
@@ -78,6 +86,47 @@ namespace ignite
break;
}
+ case OperationCallback::COMPUTE_JOB_CREATE:
+ {
+ SharedPointer<InteropMemory> mem = env->Get()->GetMemory(val);
+
+ res = env->Get()->ComputeJobCreate(mem);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_JOB_EXECUTE:
+ {
+ SharedPointer<InteropMemory> mem = env->Get()->GetMemory(val);
+
+ env->Get()->ComputeJobExecute(mem);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_JOB_DESTROY:
+ {
+ env->Get()->ComputeJobDestroy(val);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_TASK_JOB_RESULT:
+ {
+ SharedPointer<InteropMemory> mem = env->Get()->GetMemory(val);
+
+ res = env->Get()->ComputeTaskJobResult(mem);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_TASK_REDUCE:
+ {
+ env->Get()->ComputeTaskReduce(val);
+
+ break;
+ }
+
case OperationCallback::CONTINUOUS_QUERY_LISTENER_APPLY:
{
SharedPointer<InteropMemory> mem = env->Get()->GetMemory(val);
@@ -142,10 +191,32 @@ namespace ignite
long long IGNITE_CALL InLongLongLongObjectOutLong(void* target, int type, long long val1, long long val2,
long long val3, void* arg)
{
+ int64_t res = 0;
SharedPointer<IgniteEnvironment>* env = static_cast<SharedPointer<IgniteEnvironment>*>(target);
switch (type)
{
+ case OperationCallback::COMPUTE_JOB_EXECUTE_LOCAL:
+ {
+ env->Get()->ComputeJobExecuteLocal(val1);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_TASK_LOCAL_JOB_RESULT:
+ {
+ res = env->Get()->ComputeTaskLocalJobResult(val1, val2);
+
+ break;
+ }
+
+ case OperationCallback::COMPUTE_TASK_COMPLETE:
+ {
+ env->Get()->ComputeTaskComplete(val1);
+
+ break;
+ }
+
case OperationCallback::ON_START:
{
env->Get()->OnStartCallback(val1, reinterpret_cast<jobject>(arg));
@@ -168,7 +239,7 @@ namespace ignite
}
}
- return 0;
+ return res;
}
IgniteEnvironment::IgniteEnvironment(const IgniteConfiguration& cfg) :
@@ -300,6 +371,189 @@ namespace ignite
return binding;
}
+ jobject IgniteEnvironment::GetProcessorCompute(jobject proj)
+ {
+ JniErrorInfo jniErr;
+
+ jobject res = ctx.Get()->ProcessorCompute(proc.Get(), proj, &jniErr);
+
+ IgniteError err;
+
+ IgniteError::SetError(jniErr.code, jniErr.errCls, jniErr.errMsg, err);
+
+ IgniteError::ThrowIfNeeded(err);
+
+ return res;
+ }
+
+ void IgniteEnvironment::ComputeJobExecuteLocal(int64_t jobHandle)
+ {
+ SharedPointer<compute::ComputeJobHolder> job0 =
+ StaticPointerCast<compute::ComputeJobHolder>(registry.Get(jobHandle));
+
+ compute::ComputeJobHolder* job = job0.Get();
+
+ if (job)
+ job->ExecuteLocal();
+ else
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Job is not registred for handle", "jobHandle", jobHandle);
+ }
+ }
+
+ int32_t IgniteEnvironment::ComputeTaskLocalJobResult(int64_t taskHandle, int64_t jobHandle)
+ {
+ SharedPointer<compute::ComputeJobHolder> job0 =
+ StaticPointerCast<compute::ComputeJobHolder>(registry.Get(jobHandle));
+
+ compute::ComputeJobHolder* job = job0.Get();
+
+ SharedPointer<compute::ComputeTaskHolder> task0 =
+ StaticPointerCast<compute::ComputeTaskHolder>(registry.Get(taskHandle));
+
+ compute::ComputeTaskHolder* task = task0.Get();
+
+ if (task && job)
+ return task->JobResultLocal(*job);
+
+ if (!task)
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Task is not registred for handle", "taskHandle", taskHandle);
+ }
+
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Job is not registred for handle", "jobHandle", jobHandle);
+ }
+
+ void IgniteEnvironment::ComputeTaskReduce(int64_t taskHandle)
+ {
+ SharedPointer<compute::ComputeTaskHolder> task0 =
+ StaticPointerCast<compute::ComputeTaskHolder>(registry.Get(taskHandle));
+
+ compute::ComputeTaskHolder* task = task0.Get();
+
+ if (task)
+ task->Reduce();
+ else
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Task is not registred for handle", "taskHandle", taskHandle);
+ }
+ }
+
+ void IgniteEnvironment::ComputeTaskComplete(int64_t taskHandle)
+ {
+ SharedPointer<compute::ComputeTaskHolder> task0 =
+ StaticPointerCast<compute::ComputeTaskHolder>(registry.Get(taskHandle));
+
+ compute::ComputeTaskHolder* task = task0.Get();
+
+ if (task)
+ {
+ registry.Release(task->GetJobHandle());
+ registry.Release(taskHandle);
+ }
+ }
+
+ int64_t IgniteEnvironment::ComputeJobCreate(SharedPointer<InteropMemory>& mem)
+ {
+ if (!binding.Get())
+ throw IgniteError(IgniteError::IGNITE_ERR_UNKNOWN, "IgniteBinding is not initialized.");
+
+ InteropInputStream inStream(mem.Get());
+ BinaryReaderImpl reader(&inStream);
+
+ InteropOutputStream outStream(mem.Get());
+ BinaryWriterImpl writer(&outStream, GetTypeManager());
+
+ BinaryObjectImpl binJob = BinaryObjectImpl::FromMemory(*mem.Get(), inStream.Position(), 0);
+
+ int32_t jobTypeId = binJob.GetTypeId();
+
+ bool invoked = false;
+
+ int64_t handle = binding.Get()->InvokeCallback(invoked,
+ IgniteBindingImpl::CallbackType::COMPUTE_JOB_CREATE, jobTypeId, reader, writer);
+
+ if (!invoked)
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "C++ compute job is not registered on the node (did you compile your program without -rdynamic?).",
+ "jobTypeId", jobTypeId);
+ }
+
+ return handle;
+ }
+
+ void IgniteEnvironment::ComputeJobExecute(SharedPointer<InteropMemory>& mem)
+ {
+ InteropInputStream inStream(mem.Get());
+
+ InteropOutputStream outStream(mem.Get());
+ BinaryWriterImpl writer(&outStream, GetTypeManager());
+
+ int64_t jobHandle = inStream.ReadInt64();
+
+ SharedPointer<compute::ComputeJobHolder> job0 =
+ StaticPointerCast<compute::ComputeJobHolder>(registry.Get(jobHandle));
+
+ compute::ComputeJobHolder* job = job0.Get();
+
+ if (job)
+ job->ExecuteRemote(writer);
+ else
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Job is not registred for handle", "jobHandle", jobHandle);
+ }
+
+ outStream.Synchronize();
+ }
+
+ void IgniteEnvironment::ComputeJobDestroy(int64_t jobHandle)
+ {
+ registry.Release(jobHandle);
+ }
+
+ int32_t IgniteEnvironment::ComputeTaskJobResult(SharedPointer<InteropMemory>& mem)
+ {
+ InteropInputStream inStream(mem.Get());
+ BinaryReaderImpl reader(&inStream);
+
+ int64_t taskHandle = reader.ReadInt64();
+ int64_t jobHandle = reader.ReadInt64();
+
+ // Node GUID
+ reader.ReadGuid();
+
+ // Cancel flag
+ reader.ReadBool();
+
+ SharedPointer<compute::ComputeJobHolder> job0 =
+ StaticPointerCast<compute::ComputeJobHolder>(registry.Get(jobHandle));
+
+ compute::ComputeJobHolder* job = job0.Get();
+
+ SharedPointer<compute::ComputeTaskHolder> task0 =
+ StaticPointerCast<compute::ComputeTaskHolder>(registry.Get(taskHandle));
+
+ compute::ComputeTaskHolder* task = task0.Get();
+
+ if (task && job)
+ return task->JobResultRemote(*job, reader);
+
+ if (!task)
+ {
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Task is not registred for handle", "taskHandle", taskHandle);
+ }
+
+ IGNITE_ERROR_FORMATTED_1(IgniteError::IGNITE_ERR_COMPUTE_USER_UNDECLARED_EXCEPTION,
+ "Job is not registred for handle", "jobHandle", jobHandle);
+ }
+
void IgniteEnvironment::ProcessorReleaseStart()
{
if (proc.Get())
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/ignite_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/ignite_impl.cpp b/modules/platforms/cpp/core/src/impl/ignite_impl.cpp
index 546cd01..16e954c 100644
--- a/modules/platforms/cpp/core/src/impl/ignite_impl.cpp
+++ b/modules/platforms/cpp/core/src/impl/ignite_impl.cpp
@@ -59,11 +59,18 @@ namespace ignite
return env.Get()->Context();
}
- SharedPointer<IgniteBindingImpl> IgniteImpl::GetBinding()
+ IgniteImpl::SP_IgniteBindingImpl IgniteImpl::GetBinding()
{
return env.Get()->GetBinding();
}
+ IgniteImpl::SP_ComputeImpl IgniteImpl::GetCompute()
+ {
+ cluster::SP_ClusterGroupImpl serversCluster = prjImpl.Get()->ForServers();
+
+ return serversCluster.Get()->GetCompute();
+ }
+
IgniteImpl::SP_TransactionsImpl IgniteImpl::InternalGetTransactions(IgniteError &err)
{
SP_TransactionsImpl res;
@@ -80,16 +87,16 @@ namespace ignite
return res;
}
- IgniteImpl::SP_ClusterGroupImpl IgniteImpl::InternalGetProjection(IgniteError& err)
+ cluster::SP_ClusterGroupImpl IgniteImpl::InternalGetProjection(IgniteError& err)
{
- SP_ClusterGroupImpl res;
+ cluster::SP_ClusterGroupImpl res;
JniErrorInfo jniErr;
jobject txJavaRef = env.Get()->Context()->ProcessorProjection(javaRef, &jniErr);
if (txJavaRef)
- res = SP_ClusterGroupImpl(new cluster::ClusterGroupImpl(env, txJavaRef));
+ res = cluster::SP_ClusterGroupImpl(new cluster::ClusterGroupImpl(env, txJavaRef));
else
IgniteError::SetError(jniErr.code, jniErr.errCls, jniErr.errMsg, err);
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/interop/interop_target.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/interop/interop_target.cpp b/modules/platforms/cpp/core/src/impl/interop/interop_target.cpp
index b0932e7..7eed6f3 100644
--- a/modules/platforms/cpp/core/src/impl/interop/interop_target.cpp
+++ b/modules/platforms/cpp/core/src/impl/interop/interop_target.cpp
@@ -216,6 +216,26 @@ namespace ignite
return OperationResult::AI_ERROR;
}
+ jobject InteropTarget::InStreamOutObject(int32_t opType, InteropMemory& outInMem)
+ {
+ JniErrorInfo jniErr;
+
+ int64_t outInPtr = outInMem.PointerLong();
+
+ if (outInPtr)
+ {
+ jobject res = env.Get()->Context()->TargetInStreamOutObject(javaRef, opType, outInPtr, &jniErr);
+
+ IgniteError err;
+ IgniteError::SetError(jniErr.code, jniErr.errCls, jniErr.errMsg, err);
+ IgniteError::ThrowIfNeeded(err);
+
+ return res;
+ }
+
+ return 0;
+ }
+
int64_t InteropTarget::OutInOpLong(int32_t opType, int64_t val, IgniteError& err)
{
JniErrorInfo jniErr;
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/src/impl/transactions/transactions_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/src/impl/transactions/transactions_impl.cpp b/modules/platforms/cpp/core/src/impl/transactions/transactions_impl.cpp
index 133b375..b9e976a 100644
--- a/modules/platforms/cpp/core/src/impl/transactions/transactions_impl.cpp
+++ b/modules/platforms/cpp/core/src/impl/transactions/transactions_impl.cpp
@@ -165,8 +165,6 @@ namespace ignite
TransactionState::Type TransactionsImpl::TxCommit(int64_t id, IgniteError& err)
{
- JniErrorInfo jniErr;
-
int state = static_cast<int>(OutInOpLong(Operation::COMMIT, id, err));
return ToTransactionState(state);
@@ -174,8 +172,6 @@ namespace ignite
TransactionState::Type TransactionsImpl::TxRollback(int64_t id, IgniteError& err)
{
- JniErrorInfo jniErr;
-
int state = static_cast<int>(OutInOpLong(Operation::ROLLBACK, id, err));
return ToTransactionState(state);
@@ -183,8 +179,6 @@ namespace ignite
TransactionState::Type TransactionsImpl::TxClose(int64_t id, IgniteError& err)
{
- JniErrorInfo jniErr;
-
int state = static_cast<int>(OutInOpLong(Operation::CLOSE, id, err));
return ToTransactionState(state);
@@ -192,8 +186,6 @@ namespace ignite
bool TransactionsImpl::TxSetRollbackOnly(int64_t id, IgniteError& err)
{
- JniErrorInfo jniErr;
-
bool rollbackOnly = OutInOpLong(Operation::SET_ROLLBACK_ONLY, id, err) == 1;
return rollbackOnly;
@@ -201,8 +193,6 @@ namespace ignite
TransactionState::Type TransactionsImpl::TxState(int64_t id, IgniteError& err)
{
- JniErrorInfo jniErr;
-
int state = static_cast<int>(OutInOpLong(Operation::STATE, id, err));
return ToTransactionState(state);
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/jni/include/ignite/jni/java.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/jni/include/ignite/jni/java.h b/modules/platforms/cpp/jni/include/ignite/jni/java.h
index 85955b3..f6d7207 100644
--- a/modules/platforms/cpp/jni/include/ignite/jni/java.h
+++ b/modules/platforms/cpp/jni/include/ignite/jni/java.h
@@ -365,6 +365,7 @@ namespace ignite
jobject ProcessorDataStreamer(jobject obj, const char* name, bool keepPortable);
jobject ProcessorTransactions(jobject obj, JniErrorInfo* errInfo = NULL);
jobject ProcessorCompute(jobject obj, jobject prj);
+ jobject ProcessorCompute(jobject obj, jobject prj, JniErrorInfo* errInfo);
jobject ProcessorMessage(jobject obj, jobject prj);
jobject ProcessorEvents(jobject obj, jobject prj);
jobject ProcessorServices(jobject obj, jobject prj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/jni/src/java.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/jni/src/java.cpp b/modules/platforms/cpp/jni/src/java.cpp
index 809aa17..bc6af34 100644
--- a/modules/platforms/cpp/jni/src/java.cpp
+++ b/modules/platforms/cpp/jni/src/java.cpp
@@ -1154,6 +1154,16 @@ namespace ignite
return LocalToGlobal(env, res);
}
+ jobject JniContext::ProcessorCompute(jobject obj, jobject prj, JniErrorInfo* errInfo) {
+ JNIEnv* env = Attach();
+
+ jobject res = env->CallObjectMethod(obj, jvm->GetMembers().m_PlatformProcessor_compute, prj);
+
+ ExceptionCheck(env, errInfo);
+
+ return LocalToGlobal(env, res);
+ }
+
jobject JniContext::ProcessorMessage(jobject obj, jobject prj) {
JNIEnv* env = Attach();
[2/2] ignite git commit: IGNITE-3355: Implemented Compute::Call() for
C++
Posted by is...@apache.org.
IGNITE-3355: Implemented Compute::Call() for C++
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f9c96de5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f9c96de5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f9c96de5
Branch: refs/heads/master
Commit: f9c96de578c8a87e8f2f7fee01d3719d0bb21359
Parents: 0f8af13
Author: Igor Sapego <is...@gridgain.com>
Authored: Mon May 29 17:29:10 2017 +0300
Committer: Igor Sapego <is...@gridgain.com>
Committed: Mon May 29 17:29:10 2017 +0300
----------------------------------------------------------------------
modules/platforms/cpp/binary/Makefile.am | 67 ++--
.../platforms/cpp/binary/include/Makefile.am | 55 +--
.../binary/include/ignite/binary/binary_type.h | 2 +-
.../ignite/impl/binary/binary_type_impl.h | 61 ++++
.../ignite/impl/binary/binary_writer_impl.h | 4 +-
.../cpp/binary/project/vs/binary.vcxproj | 2 +
.../binary/project/vs/binary.vcxproj.filters | 6 +
.../binary/src/impl/binary/binary_type_impl.cpp | 61 ++++
modules/platforms/cpp/common/Makefile.am | 54 +--
.../platforms/cpp/common/include/Makefile.am | 1 +
.../common/include/ignite/common/cancelable.h | 65 ++++
.../cpp/common/include/ignite/common/promise.h | 18 +
.../common/include/ignite/common/shared_state.h | 51 +++
.../cpp/common/include/ignite/future.h | 60 +++-
.../cpp/common/include/ignite/ignite_error.h | 3 +
.../cpp/common/project/vs/common.vcxproj | 1 +
.../common/project/vs/common.vcxproj.filters | 3 +
modules/platforms/cpp/core-test/Makefile.am | 101 +++---
.../cpp/core-test/include/ignite/test_utils.h | 24 +-
.../cpp/core-test/project/vs/core-test.vcxproj | 1 +
.../project/vs/core-test.vcxproj.filters | 10 +-
.../cpp/core-test/src/cache_invoke_test.cpp | 2 +-
.../cpp/core-test/src/cluster_test.cpp | 2 +-
.../cpp/core-test/src/compute_test.cpp | 337 +++++++++++++++++++
.../cpp/core-test/src/continuous_query_test.cpp | 2 +-
.../platforms/cpp/core-test/src/test_utils.cpp | 9 +
modules/platforms/cpp/core/Makefile.am | 82 ++---
modules/platforms/cpp/core/include/Makefile.am | 107 +++---
.../cpp/core/include/ignite/cache/cache.h | 4 +-
.../cpp/core/include/ignite/cache/query/query.h | 1 -
.../cpp/core/include/ignite/compute/compute.h | 136 ++++++++
.../core/include/ignite/compute/compute_func.h | 65 ++++
.../platforms/cpp/core/include/ignite/ignite.h | 19 +-
.../cpp/core/include/ignite/ignite_binding.h | 25 ++
.../include/ignite/ignite_binding_context.h | 2 +-
.../cpp/core/include/ignite/impl/bindings.h | 24 ++
.../ignite/impl/cluster/cluster_group_impl.h | 31 +-
.../ignite/impl/compute/cancelable_impl.h | 70 ++++
.../include/ignite/impl/compute/compute_impl.h | 118 +++++++
.../ignite/impl/compute/compute_job_holder.h | 139 ++++++++
.../ignite/impl/compute/compute_job_result.h | 161 +++++++++
.../ignite/impl/compute/compute_task_holder.h | 213 ++++++++++++
.../include/ignite/impl/ignite_binding_impl.h | 2 +
.../include/ignite/impl/ignite_environment.h | 69 ++++
.../cpp/core/include/ignite/impl/ignite_impl.h | 23 +-
.../ignite/impl/interop/interop_target.h | 11 +-
.../platforms/cpp/core/project/vs/core.vcxproj | 9 +
.../cpp/core/project/vs/core.vcxproj.filters | 33 ++
modules/platforms/cpp/core/src/ignite.cpp | 5 +
.../src/impl/cluster/cluster_group_impl.cpp | 23 +-
.../core/src/impl/compute/cancelable_impl.cpp | 59 ++++
.../cpp/core/src/impl/compute/compute_impl.cpp | 35 ++
.../cpp/core/src/impl/ignite_environment.cpp | 260 +++++++++++++-
.../platforms/cpp/core/src/impl/ignite_impl.cpp | 15 +-
.../core/src/impl/interop/interop_target.cpp | 20 ++
.../src/impl/transactions/transactions_impl.cpp | 10 -
.../platforms/cpp/jni/include/ignite/jni/java.h | 1 +
modules/platforms/cpp/jni/src/java.cpp | 10 +
58 files changed, 2493 insertions(+), 291 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/Makefile.am b/modules/platforms/cpp/binary/Makefile.am
index d3c2072..499c8a9 100644
--- a/modules/platforms/cpp/binary/Makefile.am
+++ b/modules/platforms/cpp/binary/Makefile.am
@@ -20,51 +20,52 @@ ACLOCAL_AMFLAGS =-I m4
lib_LTLIBRARIES = libignite-binary.la
SUBDIRS = \
- include
+ include
AM_CPPFLAGS = \
- -I$(srcdir)/include \
- -I@top_srcdir@/common/include \
- -I@top_srcdir@/common/os/linux/include \
- -DIGNITE_IMPL \
- -D__STDC_LIMIT_MACROS \
- -D__STDC_CONSTANT_MACROS
+ -I$(srcdir)/include \
+ -I@top_srcdir@/common/include \
+ -I@top_srcdir@/common/os/linux/include \
+ -DIGNITE_IMPL \
+ -D__STDC_LIMIT_MACROS \
+ -D__STDC_CONSTANT_MACROS
AM_CXXFLAGS = \
- -Wall \
- -std=c++03
+ -Wall \
+ -std=c++03
libignite_binary_la_LIBADD = \
- @top_srcdir@/common/libignite-common.la
+ @top_srcdir@/common/libignite-common.la
libignite_binary_la_LDFLAGS = \
- -no-undefined \
- -version-info 0:0:0 \
- -release $(PACKAGE_VERSION)
+ -no-undefined \
+ -version-info 0:0:0 \
+ -release $(PACKAGE_VERSION)
libignite_binary_la_DEPENDENCIES = \
- @top_srcdir@/common/libignite-common.la
+ @top_srcdir@/common/libignite-common.la
libignite_binary_la_SOURCES = \
- src/binary/binary_containers.cpp \
- src/binary/binary_raw_writer.cpp \
- src/binary/binary_writer.cpp \
- src/binary/binary_reader.cpp \
- src/binary/binary_type.cpp \
- src/binary/binary_raw_reader.cpp \
- src/impl/binary/binary_type_manager.cpp \
- src/impl/binary/binary_utils.cpp \
- src/impl/binary/binary_reader_impl.cpp \
- src/impl/binary/binary_type_handler.cpp \
- src/impl/binary/binary_writer_impl.cpp \
- src/impl/binary/binary_schema.cpp \
- src/impl/binary/binary_type_snapshot.cpp \
- src/impl/binary/binary_object_header.cpp \
- src/impl/binary/binary_object_impl.cpp \
- src/impl/binary/binary_field_meta.cpp \
- src/impl/interop/interop_memory.cpp \
- src/impl/interop/interop_output_stream.cpp \
- src/impl/interop/interop_input_stream.cpp
+ src/binary/binary_containers.cpp \
+ src/binary/binary_raw_writer.cpp \
+ src/binary/binary_writer.cpp \
+ src/binary/binary_reader.cpp \
+ src/binary/binary_type.cpp \
+ src/binary/binary_raw_reader.cpp \
+ src/impl/binary/binary_type_manager.cpp \
+ src/impl/binary/binary_type_impl.cpp \
+ src/impl/binary/binary_utils.cpp \
+ src/impl/binary/binary_reader_impl.cpp \
+ src/impl/binary/binary_type_handler.cpp \
+ src/impl/binary/binary_writer_impl.cpp \
+ src/impl/binary/binary_schema.cpp \
+ src/impl/binary/binary_type_snapshot.cpp \
+ src/impl/binary/binary_object_header.cpp \
+ src/impl/binary/binary_object_impl.cpp \
+ src/impl/binary/binary_field_meta.cpp \
+ src/impl/interop/interop_memory.cpp \
+ src/impl/interop/interop_output_stream.cpp \
+ src/impl/interop/interop_input_stream.cpp
clean-local:
$(RM) *.gcno *.gcda
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/include/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/include/Makefile.am b/modules/platforms/cpp/binary/include/Makefile.am
index a73d5de..85a5516 100644
--- a/modules/platforms/cpp/binary/include/Makefile.am
+++ b/modules/platforms/cpp/binary/include/Makefile.am
@@ -18,33 +18,34 @@
ACLOCAL_AMFLAGS =-I m4
nobase_include_HEADERS = \
- ignite/binary/binary_raw_reader.h \
- ignite/binary/binary_reader.h \
- ignite/binary/binary_writer.h \
- ignite/binary/binary_raw_writer.h \
- ignite/binary/binary_containers.h \
- ignite/binary/binary.h \
- ignite/binary/binary_consts.h \
- ignite/binary/binary_type.h \
- ignite/binary/binary_object.h \
- ignite/impl/binary/binary_type_handler.h \
- ignite/impl/binary/binary_id_resolver.h \
- ignite/impl/binary/binary_type_manager.h \
- ignite/impl/binary/binary_type_updater.h \
- ignite/impl/binary/binary_common.h \
- ignite/impl/binary/binary_writer_impl.h \
- ignite/impl/binary/binary_field_meta.h \
- ignite/impl/binary/binary_type_snapshot.h \
- ignite/impl/binary/binary_reader_impl.h \
- ignite/impl/binary/binary_schema.h \
- ignite/impl/binary/binary_utils.h \
- ignite/impl/binary/binary_object_header.h \
- ignite/impl/binary/binary_object_impl.h \
- ignite/impl/interop/interop_memory.h \
- ignite/impl/interop/interop.h \
- ignite/impl/interop/interop_stream_position_guard.h \
- ignite/impl/interop/interop_output_stream.h \
- ignite/impl/interop/interop_input_stream.h
+ ignite/binary/binary_raw_reader.h \
+ ignite/binary/binary_reader.h \
+ ignite/binary/binary_writer.h \
+ ignite/binary/binary_raw_writer.h \
+ ignite/binary/binary_containers.h \
+ ignite/binary/binary.h \
+ ignite/binary/binary_consts.h \
+ ignite/binary/binary_type.h \
+ ignite/binary/binary_object.h \
+ ignite/impl/binary/binary_type_handler.h \
+ ignite/impl/binary/binary_id_resolver.h \
+ ignite/impl/binary/binary_type_impl.h \
+ ignite/impl/binary/binary_type_manager.h \
+ ignite/impl/binary/binary_type_updater.h \
+ ignite/impl/binary/binary_common.h \
+ ignite/impl/binary/binary_writer_impl.h \
+ ignite/impl/binary/binary_field_meta.h \
+ ignite/impl/binary/binary_type_snapshot.h \
+ ignite/impl/binary/binary_reader_impl.h \
+ ignite/impl/binary/binary_schema.h \
+ ignite/impl/binary/binary_utils.h \
+ ignite/impl/binary/binary_object_header.h \
+ ignite/impl/binary/binary_object_impl.h \
+ ignite/impl/interop/interop_memory.h \
+ ignite/impl/interop/interop.h \
+ ignite/impl/interop/interop_stream_position_guard.h \
+ ignite/impl/interop/interop_output_stream.h \
+ ignite/impl/interop/interop_input_stream.h
uninstall-hook:
if [ -d ${includedir}/ignite ]; then find ${includedir}/ignite -type d -empty -delete; fi
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/include/ignite/binary/binary_type.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/include/ignite/binary/binary_type.h b/modules/platforms/cpp/binary/include/ignite/binary/binary_type.h
index 8fb9dc2..749404a 100644
--- a/modules/platforms/cpp/binary/include/ignite/binary/binary_type.h
+++ b/modules/platforms/cpp/binary/include/ignite/binary/binary_type.h
@@ -28,7 +28,7 @@
#include <ignite/common/common.h>
-#include <ignite/ignite_error.h>
+#include <ignite/impl/binary/binary_type_impl.h>
/**
* @def IGNITE_BINARY_TYPE_START(T)
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_type_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_type_impl.h b/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_type_impl.h
new file mode 100644
index 0000000..2548a83
--- /dev/null
+++ b/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_type_impl.h
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#ifndef _IGNITE_IMPL_BINARY_BINARY_TYPE_IMPL
+#define _IGNITE_IMPL_BINARY_BINARY_TYPE_IMPL
+
+#include <stdint.h>
+
+#include <ignite/ignite_error.h>
+
+namespace ignite
+{
+ namespace binary
+ {
+ class BinaryReader;
+ class BinaryWriter;
+
+ template<typename T>
+ struct BinaryType;
+
+ template<>
+ struct BinaryType<IgniteError>
+ {
+ static int32_t GetTypeId();
+
+ static void GetTypeName(std::string& dst)
+ {
+ dst = "IgniteError";
+ }
+
+ static int32_t GetFieldId(const char* name);
+
+ static bool IsNull(const IgniteError& obj)
+ {
+ return false;
+ }
+
+ static void GetNull(IgniteError& dst);
+
+ static void Write(BinaryWriter& writer, const IgniteError& obj);
+
+ static void Read(BinaryReader& reader, IgniteError& dst);
+ };
+ }
+}
+
+#endif //_IGNITE_IMPL_BINARY_BINARY_TYPE_IMPL
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_writer_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_writer_impl.h b/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_writer_impl.h
index d91e2f7..32801ec 100644
--- a/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_writer_impl.h
+++ b/modules/platforms/cpp/binary/include/ignite/impl/binary/binary_writer_impl.h
@@ -59,7 +59,7 @@ namespace ignite
* @param metaMgr Type manager.
* @param metaHnd Type handler.
*/
- BinaryWriterImpl(ignite::impl::interop::InteropOutputStream* stream, BinaryIdResolver* idRslvr,
+ BinaryWriterImpl(interop::InteropOutputStream* stream, BinaryIdResolver* idRslvr,
BinaryTypeManager* metaMgr, BinaryTypeHandler* metaHnd, int32_t start);
/**
@@ -69,7 +69,7 @@ namespace ignite
* @param stream Interop stream.
* @param metaMgr Type manager.
*/
- BinaryWriterImpl(ignite::impl::interop::InteropOutputStream* stream, BinaryTypeManager* metaMgr);
+ BinaryWriterImpl(interop::InteropOutputStream* stream, BinaryTypeManager* metaMgr);
/**
* Write 8-byte signed integer. Maps to "byte" type in Java.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/project/vs/binary.vcxproj
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/project/vs/binary.vcxproj b/modules/platforms/cpp/binary/project/vs/binary.vcxproj
index 21f1248..5b76f06 100644
--- a/modules/platforms/cpp/binary/project/vs/binary.vcxproj
+++ b/modules/platforms/cpp/binary/project/vs/binary.vcxproj
@@ -194,6 +194,7 @@
<ClInclude Include="..\..\include\ignite\impl\binary\binary_reader_impl.h" />
<ClInclude Include="..\..\include\ignite\impl\binary\binary_schema.h" />
<ClInclude Include="..\..\include\ignite\impl\binary\binary_type_handler.h" />
+ <ClInclude Include="..\..\include\ignite\impl\binary\binary_type_impl.h" />
<ClInclude Include="..\..\include\ignite\impl\binary\binary_type_manager.h" />
<ClInclude Include="..\..\include\ignite\impl\binary\binary_type_snapshot.h" />
<ClInclude Include="..\..\include\ignite\impl\binary\binary_type_updater.h" />
@@ -218,6 +219,7 @@
<ClCompile Include="..\..\src\impl\binary\binary_reader_impl.cpp" />
<ClCompile Include="..\..\src\impl\binary\binary_schema.cpp" />
<ClCompile Include="..\..\src\impl\binary\binary_type_handler.cpp" />
+ <ClCompile Include="..\..\src\impl\binary\binary_type_impl.cpp" />
<ClCompile Include="..\..\src\impl\binary\binary_type_manager.cpp" />
<ClCompile Include="..\..\src\impl\binary\binary_type_snapshot.cpp" />
<ClCompile Include="..\..\src\impl\binary\binary_utils.cpp" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/project/vs/binary.vcxproj.filters
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/project/vs/binary.vcxproj.filters b/modules/platforms/cpp/binary/project/vs/binary.vcxproj.filters
index 27e4ad8..573b5fa 100644
--- a/modules/platforms/cpp/binary/project/vs/binary.vcxproj.filters
+++ b/modules/platforms/cpp/binary/project/vs/binary.vcxproj.filters
@@ -100,6 +100,9 @@
<ClInclude Include="..\..\include\ignite\impl\binary\binary_field_meta.h">
<Filter>Code\impl\binary</Filter>
</ClInclude>
+ <ClInclude Include="..\..\include\ignite\impl\binary\binary_type_impl.h">
+ <Filter>Code\impl\binary</Filter>
+ </ClInclude>
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\src\binary\binary_containers.cpp">
@@ -159,5 +162,8 @@
<ClCompile Include="..\..\src\impl\binary\binary_field_meta.cpp">
<Filter>Code\impl\binary</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\impl\binary\binary_type_impl.cpp">
+ <Filter>Code\impl\binary</Filter>
+ </ClCompile>
</ItemGroup>
</Project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/binary/src/impl/binary/binary_type_impl.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/binary/src/impl/binary/binary_type_impl.cpp b/modules/platforms/cpp/binary/src/impl/binary/binary_type_impl.cpp
new file mode 100644
index 0000000..59a4dda
--- /dev/null
+++ b/modules/platforms/cpp/binary/src/impl/binary/binary_type_impl.cpp
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#include <cstring>
+
+#include <ignite/binary/binary_reader.h>
+#include <ignite/binary/binary_writer.h>
+#include <ignite/impl/binary/binary_type_impl.h>
+
+namespace ignite
+{
+ namespace binary
+ {
+ int32_t BinaryType<IgniteError>::GetTypeId()
+ {
+ return GetBinaryStringHashCode("IgniteError");
+ }
+
+ int32_t BinaryType<IgniteError>::GetFieldId(const char* name)
+ {
+ return GetBinaryStringHashCode(name);
+ }
+
+ void BinaryType<IgniteError>::GetNull(IgniteError& dst)
+ {
+ dst = IgniteError(0, 0);
+ }
+
+ void BinaryType<IgniteError>::Write(BinaryWriter& writer, const IgniteError& obj)
+ {
+ BinaryRawWriter raw = writer.RawWriter();
+
+ raw.WriteInt32(obj.GetCode());
+ raw.WriteString(obj.GetText(), static_cast<int32_t>(strlen(obj.GetText())));
+ }
+
+ void BinaryType<IgniteError>::Read(BinaryReader& reader, IgniteError& dst)
+ {
+ BinaryRawReader raw = reader.RawReader();
+
+ int32_t code = raw.ReadInt32();
+ std::string msg = raw.ReadObject<std::string>();
+
+ dst = IgniteError(code, msg.c_str());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/Makefile.am b/modules/platforms/cpp/common/Makefile.am
index 8515d5b..374f417 100644
--- a/modules/platforms/cpp/common/Makefile.am
+++ b/modules/platforms/cpp/common/Makefile.am
@@ -20,41 +20,41 @@ ACLOCAL_AMFLAGS =-I m4
lib_LTLIBRARIES = libignite-common.la
SUBDIRS = \
- include \
- os/linux/include
+ include \
+ os/linux/include
AM_CPPFLAGS = \
- -I$(srcdir)/include \
- -I$(srcdir)/os/linux/include \
- -DIGNITE_IMPL \
- -D__STDC_LIMIT_MACROS \
- -D__STDC_CONSTANT_MACROS
+ -I$(srcdir)/include \
+ -I$(srcdir)/os/linux/include \
+ -DIGNITE_IMPL \
+ -D__STDC_LIMIT_MACROS \
+ -D__STDC_CONSTANT_MACROS
AM_CXXFLAGS = \
- -Wall \
- -std=c++03
+ -Wall \
+ -std=c++03
libignite_common_la_LDFLAGS = \
- -no-undefined \
- -L/usr/local/lib \
- -ldl \
- -version-info 0:0:0 \
- -release $(PACKAGE_VERSION)
+ -no-undefined \
+ -L/usr/local/lib \
+ -ldl \
+ -version-info 0:0:0 \
+ -release $(PACKAGE_VERSION)
libignite_common_la_SOURCES = \
- os/linux/src/common/concurrent_os.cpp \
- os/linux/src/common/platform_utils.cpp \
- os/linux/src/common/dynamic_load_os.cpp \
- src/common/big_integer.cpp \
- src/common/concurrent.cpp \
- src/common/decimal.cpp \
- src/common/bits.cpp \
- src/common/utils.cpp \
- src/date.cpp \
- src/ignite_error.cpp \
- src/guid.cpp \
- src/timestamp.cpp \
- src/time.cpp
+ os/linux/src/common/concurrent_os.cpp \
+ os/linux/src/common/platform_utils.cpp \
+ os/linux/src/common/dynamic_load_os.cpp \
+ src/common/big_integer.cpp \
+ src/common/concurrent.cpp \
+ src/common/decimal.cpp \
+ src/common/bits.cpp \
+ src/common/utils.cpp \
+ src/date.cpp \
+ src/ignite_error.cpp \
+ src/guid.cpp \
+ src/timestamp.cpp \
+ src/time.cpp
clean-local:
$(RM) *.gcno *.gcda
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/Makefile.am b/modules/platforms/cpp/common/include/Makefile.am
index 64f0c46..f115c94 100644
--- a/modules/platforms/cpp/common/include/Makefile.am
+++ b/modules/platforms/cpp/common/include/Makefile.am
@@ -20,6 +20,7 @@ ACLOCAL_AMFLAGS =-I m4
nobase_include_HEADERS = \
ignite/common/big_integer.h \
ignite/common/bits.h \
+ ignite/common/cancelable.h \
ignite/common/concurrent.h \
ignite/common/decimal.h \
ignite/common/default_allocator.h \
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/ignite/common/cancelable.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/cancelable.h b/modules/platforms/cpp/common/include/ignite/common/cancelable.h
new file mode 100644
index 0000000..ab0c126
--- /dev/null
+++ b/modules/platforms/cpp/common/include/ignite/common/cancelable.h
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::common::Cancelable class.
+ */
+
+#ifndef _IGNITE_COMMON_CANCELABLE
+#define _IGNITE_COMMON_CANCELABLE
+
+#include <ignite/common/common.h>
+
+namespace ignite
+{
+ namespace common
+ {
+ /**
+ * Cancelable.
+ */
+ class IGNITE_IMPORT_EXPORT Cancelable
+ {
+ public:
+ /**
+ * Default constructor.
+ */
+ Cancelable()
+ {
+ // No-op.
+ }
+
+ /**
+ * Destructor.
+ */
+ virtual ~Cancelable()
+ {
+ // No-op.
+ }
+
+ /**
+ * Cancels the operation.
+ */
+ virtual void Cancel() = 0;
+
+ private:
+ IGNITE_NO_COPY_ASSIGNMENT(Cancelable);
+ };
+ }
+}
+
+#endif //_IGNITE_COMMON_CANCELABLE
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/ignite/common/promise.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/promise.h b/modules/platforms/cpp/common/include/ignite/common/promise.h
index 548b76b..2e6152f 100644
--- a/modules/platforms/cpp/common/include/ignite/common/promise.h
+++ b/modules/platforms/cpp/common/include/ignite/common/promise.h
@@ -111,9 +111,18 @@ namespace ignite
state.Get()->SetError(err);
}
+ /**
+ * Set cancel target.
+ */
+ void SetCancelTarget(std::auto_ptr<Cancelable>& target)
+ {
+ state.Get()->SetCancelTarget(target);
+ }
+
private:
IGNITE_NO_COPY_ASSIGNMENT(Promise);
+ /** Shared state. */
concurrent::SharedPointer< SharedState<ValueType> > state;
};
@@ -190,9 +199,18 @@ namespace ignite
state.Get()->SetError(err);
}
+ /**
+ * Set cancel target.
+ */
+ void SetCancelTarget(std::auto_ptr<Cancelable>& target)
+ {
+ state.Get()->SetCancelTarget(target);
+ }
+
private:
IGNITE_NO_COPY_ASSIGNMENT(Promise);
+ /** Shared state. */
concurrent::SharedPointer< SharedState<ValueType> > state;
};
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/ignite/common/shared_state.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/shared_state.h b/modules/platforms/cpp/common/include/ignite/common/shared_state.h
index 8886532..d223753 100644
--- a/modules/platforms/cpp/common/include/ignite/common/shared_state.h
+++ b/modules/platforms/cpp/common/include/ignite/common/shared_state.h
@@ -23,8 +23,11 @@
#ifndef _IGNITE_SHARED_STATE
#define _IGNITE_SHARED_STATE
+#include <memory>
+
#include <ignite/common/common.h>
#include <ignite/common/concurrent.h>
+#include <ignite/common/cancelable.h>
#include <ignite/ignite_error.h>
namespace ignite
@@ -162,9 +165,33 @@ namespace ignite
throw error;
}
+ /**
+ * Set cancel target.
+ */
+ void SetCancelTarget(std::auto_ptr<Cancelable>& target)
+ {
+ concurrent::CsLockGuard guard(mutex);
+
+ cancelTarget = target;
+ }
+
+ /**
+ * Cancel related operation.
+ */
+ void Cancel()
+ {
+ concurrent::CsLockGuard guard(mutex);
+
+ if (cancelTarget.get())
+ cancelTarget->Cancel();
+ }
+
private:
IGNITE_NO_COPY_ASSIGNMENT(SharedState);
+ /** Cancel target. */
+ std::auto_ptr<Cancelable> cancelTarget;
+
/** Value. */
std::auto_ptr<ValueType> value;
@@ -310,9 +337,33 @@ namespace ignite
throw error;
}
+ /**
+ * Set cancel target.
+ */
+ void SetCancelTarget(std::auto_ptr<Cancelable>& target)
+ {
+ concurrent::CsLockGuard guard(mutex);
+
+ cancelTarget = target;
+ }
+
+ /**
+ * Cancel related operation.
+ */
+ void Cancel()
+ {
+ concurrent::CsLockGuard guard(mutex);
+
+ if (cancelTarget.get())
+ cancelTarget->Cancel();
+ }
+
private:
IGNITE_NO_COPY_ASSIGNMENT(SharedState);
+ /** Cancel target. */
+ std::auto_ptr<Cancelable> cancelTarget;
+
/** Marker. */
bool done;
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/ignite/future.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/future.h b/modules/platforms/cpp/common/include/ignite/future.h
index 5c42e55..f709797 100644
--- a/modules/platforms/cpp/common/include/ignite/future.h
+++ b/modules/platforms/cpp/common/include/ignite/future.h
@@ -85,7 +85,7 @@ namespace ignite
assert(state0 != 0);
- state.Get()->Wait();
+ state0->Wait();
}
/**
@@ -101,7 +101,7 @@ namespace ignite
assert(state0 != 0);
- return state.Get()->WaitFor(msTimeout);
+ return state0->WaitFor(msTimeout);
}
/**
@@ -117,7 +117,31 @@ namespace ignite
assert(state0 != 0);
- return state.Get()->GetValue();
+ return state0->GetValue();
+ }
+
+ /**
+ * Cancel related operation.
+ */
+ void Cancel()
+ {
+ common::SharedState<ValueType>* state0 = state.Get();
+
+ assert(state0 != 0);
+
+ state0->Cancel();
+ }
+
+ /**
+ * Check if the future ready.
+ */
+ bool IsReady()
+ {
+ common::SharedState<ValueType>* state0 = state.Get();
+
+ assert(state0 != 0);
+
+ return state0->IsSet();
}
private:
@@ -182,7 +206,7 @@ namespace ignite
assert(state0 != 0);
- state.Get()->Wait();
+ state0->Wait();
}
/**
@@ -198,7 +222,7 @@ namespace ignite
assert(state0 != 0);
- return state.Get()->WaitFor(msTimeout);
+ return state0->WaitFor(msTimeout);
}
/**
@@ -213,7 +237,31 @@ namespace ignite
assert(state0 != 0);
- state.Get()->GetValue();
+ state0->GetValue();
+ }
+
+ /**
+ * Cancel related operation.
+ */
+ void Cancel()
+ {
+ common::SharedState<ValueType>* state0 = state.Get();
+
+ assert(state0 != 0);
+
+ state0->Cancel();
+ }
+
+ /**
+ * Check if the future ready.
+ */
+ bool IsReady()
+ {
+ common::SharedState<ValueType>* state0 = state.Get();
+
+ assert(state0 != 0);
+
+ return state0->IsSet();
}
private:
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/include/ignite/ignite_error.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/ignite_error.h b/modules/platforms/cpp/common/include/ignite/ignite_error.h
index 7818dd2..59bb0c4 100644
--- a/modules/platforms/cpp/common/include/ignite/ignite_error.h
+++ b/modules/platforms/cpp/common/include/ignite/ignite_error.h
@@ -124,6 +124,9 @@ namespace ignite
/** Binary error. */
static const int IGNITE_ERR_BINARY = 1002;
+ /** Standard library exception. */
+ static const int IGNITE_ERR_STD = 1003;
+
/** Generic %Ignite error. */
static const int IGNITE_ERR_GENERIC = 2000;
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/project/vs/common.vcxproj
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/project/vs/common.vcxproj b/modules/platforms/cpp/common/project/vs/common.vcxproj
index 9f32461..07c3f50 100644
--- a/modules/platforms/cpp/common/project/vs/common.vcxproj
+++ b/modules/platforms/cpp/common/project/vs/common.vcxproj
@@ -166,6 +166,7 @@
</ItemDefinitionGroup>
<ItemGroup>
<ClInclude Include="..\..\include\ignite\common\big_integer.h" />
+ <ClInclude Include="..\..\include\ignite\common\cancelable.h" />
<ClInclude Include="..\..\include\ignite\common\concurrent.h" />
<ClInclude Include="..\..\include\ignite\common\decimal.h" />
<ClInclude Include="..\..\include\ignite\common\default_allocator.h" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/common/project/vs/common.vcxproj.filters
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/project/vs/common.vcxproj.filters b/modules/platforms/cpp/common/project/vs/common.vcxproj.filters
index 4daf3aa..020d32b 100644
--- a/modules/platforms/cpp/common/project/vs/common.vcxproj.filters
+++ b/modules/platforms/cpp/common/project/vs/common.vcxproj.filters
@@ -76,6 +76,9 @@
<ClInclude Include="..\..\include\ignite\common\shared_state.h">
<Filter>Code\common</Filter>
</ClInclude>
+ <ClInclude Include="..\..\include\ignite\common\cancelable.h">
+ <Filter>Code\common</Filter>
+ </ClInclude>
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\src\date.cpp">
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/Makefile.am b/modules/platforms/cpp/core-test/Makefile.am
index 11f4d1a..f87effd 100644
--- a/modules/platforms/cpp/core-test/Makefile.am
+++ b/modules/platforms/cpp/core-test/Makefile.am
@@ -20,67 +20,68 @@ ACLOCAL_AMFLAGS =-I m4
noinst_PROGRAMS = ignite-tests
SUBDIRS = \
- include
+ include
AM_CPPFLAGS = \
- -I$(srcdir)/include \
- -I@top_srcdir@/core/include \
- -I@top_srcdir@/core/os/linux/include \
- -I@top_srcdir@/common/include \
- -I@top_srcdir@/common/os/linux/include \
- -I@top_srcdir@/binary/include \
- -I@top_srcdir@/jni/include \
- -I@top_srcdir@/jni/os/linux/include \
- -I$(JAVA_HOME)/include \
- -I$(JAVA_HOME)/include/linux \
- -DIGNITE_IMPL \
- -D__STDC_LIMIT_MACROS \
- -D__STDC_CONSTANT_MACROS
+ -I$(srcdir)/include \
+ -I@top_srcdir@/core/include \
+ -I@top_srcdir@/core/os/linux/include \
+ -I@top_srcdir@/common/include \
+ -I@top_srcdir@/common/os/linux/include \
+ -I@top_srcdir@/binary/include \
+ -I@top_srcdir@/jni/include \
+ -I@top_srcdir@/jni/os/linux/include \
+ -I$(JAVA_HOME)/include \
+ -I$(JAVA_HOME)/include/linux \
+ -DIGNITE_IMPL \
+ -D__STDC_LIMIT_MACROS \
+ -D__STDC_CONSTANT_MACROS
AM_CXXFLAGS = \
- -Wall \
- -std=c++03
+ -Wall \
+ -std=c++03
ignite_tests_LDADD = \
- @top_srcdir@/core/libignite.la \
- -lpthread \
- -lboost_thread \
- -lboost_system \
- -lboost_chrono
+ @top_srcdir@/core/libignite.la \
+ -lpthread \
+ -lboost_thread \
+ -lboost_system \
+ -lboost_chrono
ignite_tests_LDFLAGS = \
- -static-libtool-libs \
- -rdynamic
+ -static-libtool-libs \
+ -rdynamic
ignite_tests_SOURCES = \
- src/reference_test.cpp \
- src/bits_test.cpp \
+ src/reference_test.cpp \
+ src/bits_test.cpp \
src/binary_identity_resolver_test.cpp \
- src/cache_test.cpp \
- src/cache_query_test.cpp \
- src/cache_store_test.cpp \
- src/continuous_query_test.cpp \
- src/concurrent_test.cpp \
- src/ignition_test.cpp \
- src/interop_memory_test.cpp \
- src/interop_test.cpp \
- src/cluster_test.cpp \
- src/cache_invoke_test.cpp \
- src/handle_registry_test.cpp \
- src/ignite_error_test.cpp \
- src/binary_test_defs.cpp \
- src/binary_object_test.cpp \
- src/binary_reader_writer_raw_test.cpp \
- src/binary_reader_writer_test.cpp \
- src/binary_session_test.cpp \
- src/date_time_test.cpp \
- src/decimal_test.cpp \
- src/dynamic_size_array_test.cpp \
- src/fixed_size_array_test.cpp \
- src/future_test.cpp \
- src/transactions_test.cpp \
- src/teamcity_messages.cpp \
- src/teamcity_boost.cpp \
+ src/cache_test.cpp \
+ src/cache_query_test.cpp \
+ src/cache_store_test.cpp \
+ src/continuous_query_test.cpp \
+ src/concurrent_test.cpp \
+ src/compute_test.cpp \
+ src/ignition_test.cpp \
+ src/interop_memory_test.cpp \
+ src/interop_test.cpp \
+ src/cluster_test.cpp \
+ src/cache_invoke_test.cpp \
+ src/handle_registry_test.cpp \
+ src/ignite_error_test.cpp \
+ src/binary_test_defs.cpp \
+ src/binary_object_test.cpp \
+ src/binary_reader_writer_raw_test.cpp \
+ src/binary_reader_writer_test.cpp \
+ src/binary_session_test.cpp \
+ src/date_time_test.cpp \
+ src/decimal_test.cpp \
+ src/dynamic_size_array_test.cpp \
+ src/fixed_size_array_test.cpp \
+ src/future_test.cpp \
+ src/transactions_test.cpp \
+ src/teamcity_messages.cpp \
+ src/teamcity_boost.cpp \
src/test_utils.cpp
run-check: check
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/include/ignite/test_utils.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/include/ignite/test_utils.h b/modules/platforms/cpp/core-test/include/ignite/test_utils.h
index 67fda95..aa91458 100644
--- a/modules/platforms/cpp/core-test/include/ignite/test_utils.h
+++ b/modules/platforms/cpp/core-test/include/ignite/test_utils.h
@@ -22,6 +22,11 @@
namespace ignite_test
{
+ enum
+ {
+ TEST_ERROR = 424242
+ };
+
/**
* Initialize configuration for a node.
*
@@ -65,9 +70,24 @@ namespace ignite_test
* @param err Error.
* @return True if the error is generic.
*/
- inline bool IsGenericError(const ignite::IgniteError& err)
+ bool IsGenericError(const ignite::IgniteError& err);
+
+ /**
+ * Check if the error is generic.
+ *
+ * @param err Error.
+ * @return True if the error is generic.
+ */
+ bool IsTestError(const ignite::IgniteError& err);
+
+ /**
+ * Make test error.
+ *
+ * @return Test error.
+ */
+ inline ignite::IgniteError MakeTestError()
{
- return err.GetCode() == ignite::IgniteError::IGNITE_ERR_GENERIC;
+ return ignite::IgniteError(TEST_ERROR, "Test error");
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj b/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj
index 51ae5d41..f2f6a1b 100644
--- a/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj
+++ b/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj
@@ -64,6 +64,7 @@
<ClCompile Include="..\..\src\cache_test.cpp" />
<ClCompile Include="..\..\src\cluster_test.cpp" />
<ClCompile Include="..\..\src\cache_invoke_test.cpp" />
+ <ClCompile Include="..\..\src\compute_test.cpp" />
<ClCompile Include="..\..\src\concurrent_test.cpp" />
<ClCompile Include="..\..\src\date_time_test.cpp" />
<ClCompile Include="..\..\src\decimal_test.cpp" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj.filters
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj.filters b/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj.filters
index ebccc7f..d58ef97 100644
--- a/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj.filters
+++ b/modules/platforms/cpp/core-test/project/vs/core-test.vcxproj.filters
@@ -88,6 +88,9 @@
<ClCompile Include="..\..\src\future_test.cpp">
<Filter>Code</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\compute_test.cpp">
+ <Filter>Code</Filter>
+ </ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\..\include\teamcity_messages.h">
@@ -166,9 +169,6 @@
<None Include="..\..\config\cache-identity-default.xml">
<Filter>Configs</Filter>
</None>
- <None Include="..\..\config\cache-query-continuous.xml">
- <Filter>Configs</Filter>
- </None>
<None Include="..\..\config\cache-store-default.xml">
<Filter>Configs</Filter>
</None>
@@ -178,9 +178,5 @@
<None Include="..\..\config\cache-store-32.xml">
<Filter>Configs</Filter>
</None>
- <None Include="..\..\config\cache-query-continuous.xml">
- <Filter>Configs</Filter>
- </None>
- <None Include="..\..\config\cache-query-continuous.xml" />
</ItemGroup>
</Project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/src/cache_invoke_test.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/src/cache_invoke_test.cpp b/modules/platforms/cpp/core-test/src/cache_invoke_test.cpp
index 0c30af8..55fca85 100644
--- a/modules/platforms/cpp/core-test/src/cache_invoke_test.cpp
+++ b/modules/platforms/cpp/core-test/src/cache_invoke_test.cpp
@@ -379,7 +379,7 @@ namespace ignite
IGNITE_EXPORTED_CALL void IgniteModuleInit(ignite::IgniteBindingContext& context)
{
- IgniteBinding binding = context.GetBingding();
+ IgniteBinding binding = context.GetBinding();
binding.RegisterCacheEntryProcessor<CacheEntryModifier>();
binding.RegisterCacheEntryProcessor<Divisor>();
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/src/cluster_test.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/src/cluster_test.cpp b/modules/platforms/cpp/core-test/src/cluster_test.cpp
index 754fe7e..e9d6728 100644
--- a/modules/platforms/cpp/core-test/src/cluster_test.cpp
+++ b/modules/platforms/cpp/core-test/src/cluster_test.cpp
@@ -80,7 +80,7 @@ BOOST_AUTO_TEST_CASE(IgniteImplForServers)
IgniteError err;
- BOOST_REQUIRE(clusterGroup.Get()->ForServers(err).IsValid());
+ BOOST_REQUIRE(clusterGroup.Get()->ForServers().IsValid());
}
BOOST_AUTO_TEST_SUITE_END()
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/src/compute_test.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/src/compute_test.cpp b/modules/platforms/cpp/core-test/src/compute_test.cpp
new file mode 100644
index 0000000..d3b1183
--- /dev/null
+++ b/modules/platforms/cpp/core-test/src/compute_test.cpp
@@ -0,0 +1,337 @@
+/*
+ * 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.
+ */
+
+#ifndef _MSC_VER
+ #define BOOST_TEST_DYN_LINK
+#endif
+
+#include <boost/test/unit_test.hpp>
+#include <boost/chrono.hpp>
+#include <boost/thread.hpp>
+
+#include <ignite/ignition.h>
+#include <ignite/test_utils.h>
+
+#include <ignite/test_utils.h>
+
+using namespace ignite;
+using namespace ignite::compute;
+using namespace ignite::common::concurrent;
+using namespace ignite_test;
+
+using namespace boost::unit_test;
+
+/*
+ * Test setup fixture.
+ */
+struct ComputeTestSuiteFixture
+{
+ Ignite node;
+
+ Ignite MakeNode(const char* name)
+ {
+#ifdef IGNITE_TESTS_32
+ const char* config = "cache-test-32.xml";
+#else
+ const char* config = "cache-test.xml";
+#endif
+ return StartNode(config, name);
+ }
+
+ /*
+ * Constructor.
+ */
+ ComputeTestSuiteFixture() :
+ node(MakeNode("ComputeNode1"))
+ {
+ // No-op.
+ }
+
+ /*
+ * Destructor.
+ */
+ ~ComputeTestSuiteFixture()
+ {
+ Ignition::StopAll(true);
+ }
+};
+
+struct Func1 : ComputeFunc<std::string>
+{
+ Func1() :
+ a(), b(), err()
+ {
+ // No-op.
+ }
+
+ Func1(int32_t a, int32_t b) :
+ a(a), b(b), err()
+ {
+ // No-op.
+ }
+
+ Func1(IgniteError err) :
+ a(), b(), err(err)
+ {
+ // No-op.
+ }
+
+ virtual std::string Call()
+ {
+ if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
+ throw err;
+
+ std::stringstream tmp;
+
+ tmp << a << '.' << b;
+
+ return tmp.str();
+ }
+
+ int32_t a;
+ int32_t b;
+ IgniteError err;
+};
+
+struct Func2 : ComputeFunc<std::string>
+{
+ Func2() :
+ a(), b(), err()
+ {
+ // No-op.
+ }
+
+ Func2(int32_t a, int32_t b) :
+ a(a), b(b), err()
+ {
+ // No-op.
+ }
+
+ Func2(IgniteError err) :
+ a(), b(), err(err)
+ {
+ // No-op.
+ }
+
+ virtual std::string Call()
+ {
+ boost::this_thread::sleep_for(boost::chrono::milliseconds(200));
+
+ if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
+ throw err;
+
+ std::stringstream tmp;
+
+ tmp << a << '.' << b;
+
+ return tmp.str();
+ }
+
+ int32_t a;
+ int32_t b;
+ IgniteError err;
+};
+
+namespace ignite
+{
+ namespace binary
+ {
+ template<>
+ struct BinaryType<Func1>
+ {
+ static int32_t GetTypeId()
+ {
+ return GetBinaryStringHashCode("Func1");
+ }
+
+ static void GetTypeName(std::string& dst)
+ {
+ dst = "Func1";
+ }
+
+ static int32_t GetFieldId(const char* name)
+ {
+ return GetBinaryStringHashCode(name);
+ }
+
+ static bool IsNull(const Func1& obj)
+ {
+ return false;
+ }
+
+ static void GetNull(Func1& dst)
+ {
+ dst = Func1(0, 0);
+ }
+
+ static void Write(BinaryWriter& writer, const Func1& obj)
+ {
+ writer.WriteInt32("a", obj.a);
+ writer.WriteInt32("b", obj.b);
+ writer.WriteObject<IgniteError>("err", obj.err);
+ }
+
+ static void Read(BinaryReader& reader, Func1& dst)
+ {
+ dst.a = reader.ReadInt32("a");
+ dst.b = reader.ReadInt32("b");
+ dst.err = reader.ReadObject<IgniteError>("err");
+ }
+ };
+
+ template<>
+ struct BinaryType<Func2>
+ {
+ static int32_t GetTypeId()
+ {
+ return GetBinaryStringHashCode("Func2");
+ }
+
+ static void GetTypeName(std::string& dst)
+ {
+ dst = "Func2";
+ }
+
+ static int32_t GetFieldId(const char* name)
+ {
+ return GetBinaryStringHashCode(name);
+ }
+
+ static bool IsNull(const Func2& obj)
+ {
+ return false;
+ }
+
+ static void GetNull(Func2& dst)
+ {
+ dst = Func2(0, 0);
+ }
+
+ static void Write(BinaryWriter& writer, const Func2& obj)
+ {
+ writer.WriteInt32("a", obj.a);
+ writer.WriteInt32("b", obj.b);
+ writer.WriteObject<IgniteError>("err", obj.err);
+ }
+
+ static void Read(BinaryReader& reader, Func2& dst)
+ {
+ dst.a = reader.ReadInt32("a");
+ dst.b = reader.ReadInt32("b");
+ dst.err = reader.ReadObject<IgniteError>("err");
+ }
+ };
+ }
+}
+
+IGNITE_EXPORTED_CALL void IgniteModuleInit1(IgniteBindingContext& context)
+{
+ IgniteBinding binding = context.GetBinding();
+
+ binding.RegisterComputeFunc<Func1>();
+ binding.RegisterComputeFunc<Func2>();
+}
+
+BOOST_FIXTURE_TEST_SUITE(ComputeTestSuite, ComputeTestSuiteFixture)
+
+BOOST_AUTO_TEST_CASE(IgniteCallSyncLocal)
+{
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+ std::string res = compute.Call<std::string>(Func1(8, 5));
+
+ BOOST_CHECK_EQUAL(res, "8.5");
+}
+
+BOOST_AUTO_TEST_CASE(IgniteCallAsyncLocal)
+{
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+ Future<std::string> res = compute.CallAsync<std::string>(Func2(312, 245));
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECKPOINT("Waiting with timeout");
+ res.WaitFor(100);
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECK_EQUAL(res.GetValue(), "312.245");
+}
+
+BOOST_AUTO_TEST_CASE(IgniteCallSyncLocalError)
+{
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+
+ BOOST_CHECK_EXCEPTION(compute.Call<std::string>(Func1(MakeTestError())), IgniteError, IsTestError);
+}
+
+BOOST_AUTO_TEST_CASE(IgniteCallAsyncLocalError)
+{
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+ Future<std::string> res = compute.CallAsync<std::string>(Func2(MakeTestError()));
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECKPOINT("Waiting with timeout");
+ res.WaitFor(100);
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
+}
+
+BOOST_AUTO_TEST_CASE(IgniteCallTestRemote)
+{
+ Ignite node2 = MakeNode("ComputeNode2");
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+ compute.CallAsync<std::string>(Func2(8, 5));
+
+ std::string res = compute.Call<std::string>(Func1(42, 24));
+
+ BOOST_CHECK_EQUAL(res, "42.24");
+}
+
+BOOST_AUTO_TEST_CASE(IgniteCallTestRemoteError)
+{
+ Ignite node2 = MakeNode("ComputeNode2");
+ Compute compute = node.GetCompute();
+
+ BOOST_CHECKPOINT("Making Call");
+ compute.CallAsync<std::string>(Func2(8, 5));
+
+ Future<std::string> res = compute.CallAsync<std::string>(Func2(MakeTestError()));
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECKPOINT("Waiting with timeout");
+ res.WaitFor(100);
+
+ BOOST_CHECK(!res.IsReady());
+
+ BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
+}
+
+BOOST_AUTO_TEST_SUITE_END()
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/src/continuous_query_test.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/src/continuous_query_test.cpp b/modules/platforms/cpp/core-test/src/continuous_query_test.cpp
index e092728..b869eb6 100644
--- a/modules/platforms/cpp/core-test/src/continuous_query_test.cpp
+++ b/modules/platforms/cpp/core-test/src/continuous_query_test.cpp
@@ -371,7 +371,7 @@ void CheckEvents(Cache<int, TestEntry>& cache, Listener<int, TestEntry>& lsnr)
IGNITE_EXPORTED_CALL void IgniteModuleInit0(ignite::IgniteBindingContext& context)
{
- IgniteBinding binding = context.GetBingding();
+ IgniteBinding binding = context.GetBinding();
binding.RegisterCacheEntryEventFilter< RangeFilter<int, TestEntry> >();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core-test/src/test_utils.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core-test/src/test_utils.cpp b/modules/platforms/cpp/core-test/src/test_utils.cpp
index 7323028..c6e4454 100644
--- a/modules/platforms/cpp/core-test/src/test_utils.cpp
+++ b/modules/platforms/cpp/core-test/src/test_utils.cpp
@@ -83,4 +83,13 @@ namespace ignite_test
return Ignition::Start(cfg, name);
}
+ bool IsGenericError(const ignite::IgniteError& err)
+ {
+ return err.GetCode() == ignite::IgniteError::IGNITE_ERR_GENERIC;
+ }
+
+ bool IsTestError(const ignite::IgniteError& err)
+ {
+ return err.GetCode() == TEST_ERROR;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/Makefile.am b/modules/platforms/cpp/core/Makefile.am
index 4de45d3..1c970e8 100644
--- a/modules/platforms/cpp/core/Makefile.am
+++ b/modules/platforms/cpp/core/Makefile.am
@@ -20,58 +20,60 @@ ACLOCAL_AMFLAGS =-I m4
lib_LTLIBRARIES = libignite.la
SUBDIRS = \
- include
+ include
AM_CPPFLAGS = \
- -I$(srcdir)/include \
- -I@top_srcdir@/common/include \
- -I@top_srcdir@/common/os/linux/include \
- -I@top_srcdir@/binary/include \
- -I@top_srcdir@/jni/include \
- -I@top_srcdir@/jni/os/linux/include \
- -I$(JAVA_HOME)/include \
- -I$(JAVA_HOME)/include/linux \
- -DIGNITE_IMPL \
- -D__STDC_LIMIT_MACROS \
- -D__STDC_CONSTANT_MACROS
+ -I$(srcdir)/include \
+ -I@top_srcdir@/common/include \
+ -I@top_srcdir@/common/os/linux/include \
+ -I@top_srcdir@/binary/include \
+ -I@top_srcdir@/jni/include \
+ -I@top_srcdir@/jni/os/linux/include \
+ -I$(JAVA_HOME)/include \
+ -I$(JAVA_HOME)/include/linux \
+ -DIGNITE_IMPL \
+ -D__STDC_LIMIT_MACROS \
+ -D__STDC_CONSTANT_MACROS
AM_CXXFLAGS = \
- -Wall \
- -std=c++03
+ -Wall \
+ -std=c++03
libignite_la_LIBADD = \
- @top_srcdir@/jni/libignite-jni.la \
- @top_srcdir@/binary/libignite-binary.la
+ @top_srcdir@/jni/libignite-jni.la \
+ @top_srcdir@/binary/libignite-binary.la
libignite_la_LDFLAGS = \
- -no-undefined \
- -ldl \
- -version-info 0:0:0 \
- -release $(PACKAGE_VERSION)
+ -no-undefined \
+ -ldl \
+ -version-info 0:0:0 \
+ -release $(PACKAGE_VERSION)
libignite_la_DEPENDENCIES = \
- @top_srcdir@/jni/libignite-jni.la \
- @top_srcdir@/binary/libignite-binary.la
+ @top_srcdir@/jni/libignite-jni.la \
+ @top_srcdir@/binary/libignite-binary.la
libignite_la_SOURCES = \
- src/ignite.cpp \
- src/ignition.cpp \
- src/impl/ignite_environment.cpp \
- src/impl/binary/binary_type_updater_impl.cpp \
- src/impl/handle_registry.cpp \
- src/impl/cache/query/continuous/continuous_query_handle_impl.cpp \
- src/impl/cache/query/query_impl.cpp \
- src/impl/cache/cache_impl.cpp \
- src/impl/cache/query/query_batch.cpp \
- src/impl/interop/interop_external_memory.cpp \
- src/impl/interop/interop_target.cpp \
- src/impl/transactions/transaction_impl.cpp \
- src/impl/transactions/transactions_impl.cpp \
- src/impl/cluster/cluster_group_impl.cpp \
- src/impl/ignite_impl.cpp \
- src/impl/ignite_binding_impl.cpp \
- src/transactions/transaction.cpp \
- src/transactions/transactions.cpp
+ src/ignite.cpp \
+ src/ignition.cpp \
+ src/impl/ignite_environment.cpp \
+ src/impl/binary/binary_type_updater_impl.cpp \
+ src/impl/handle_registry.cpp \
+ src/impl/cache/query/continuous/continuous_query_handle_impl.cpp \
+ src/impl/cache/query/query_impl.cpp \
+ src/impl/cache/cache_impl.cpp \
+ src/impl/cache/query/query_batch.cpp \
+ src/impl/interop/interop_external_memory.cpp \
+ src/impl/interop/interop_target.cpp \
+ src/impl/transactions/transaction_impl.cpp \
+ src/impl/transactions/transactions_impl.cpp \
+ src/impl/cluster/cluster_group_impl.cpp \
+ src/impl/compute/cancelable_impl.cpp \
+ src/impl/compute/compute_impl.cpp \
+ src/impl/ignite_impl.cpp \
+ src/impl/ignite_binding_impl.cpp \
+ src/transactions/transaction.cpp \
+ src/transactions/transactions.cpp
pkgconfigdir = $(libdir)/pkgconfig
pkgconfig_DATA = $(srcdir)/ignite.pc
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/Makefile.am
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/Makefile.am b/modules/platforms/cpp/core/include/Makefile.am
index 0e9a7ec..50772cb 100644
--- a/modules/platforms/cpp/core/include/Makefile.am
+++ b/modules/platforms/cpp/core/include/Makefile.am
@@ -18,56 +18,63 @@
ACLOCAL_AMFLAGS =-I m4
nobase_include_HEADERS = \
- ignite/cache/cache.h \
- ignite/cache/cache_entry.h \
- ignite/cache/cache_entry_processor.h \
- ignite/cache/cache_peek_mode.h \
- ignite/cache/event/cache_entry_event.h \
- ignite/cache/event/cache_entry_event_filter.h \
- ignite/cache/event/cache_entry_event_listener.h \
- ignite/cache/mutable_cache_entry.h \
- ignite/cache/query/continuous/continuous_query.h \
- ignite/cache/query/continuous/continuous_query_handle.h \
- ignite/cache/query/query.h \
- ignite/cache/query/query_cursor.h \
- ignite/cache/query/query_fields_cursor.h \
- ignite/cache/query/query_fields_row.h \
- ignite/cache/query/query_scan.h \
- ignite/cache/query/query_sql.h \
- ignite/cache/query/query_sql_fields.h \
- ignite/cache/query/query_text.h \
- ignite/ignite.h \
- ignite/ignite_binding.h \
- ignite/ignite_binding_context.h \
- ignite/ignite_configuration.h \
- ignite/ignition.h \
- ignite/impl/binary/binary_type_updater_impl.h \
- ignite/impl/bindings.h \
- ignite/impl/cache/cache_entry_processor_holder.h \
- ignite/impl/cache/cache_impl.h \
- ignite/impl/cache/event/cache_entry_event_filter_base.h \
- ignite/impl/cache/event/cache_entry_event_filter_holder.h \
- ignite/impl/cache/query/continuous/continuous_query_handle_impl.h \
- ignite/impl/cache/query/continuous/continuous_query_impl.h \
- ignite/impl/cache/query/query_argument.h \
- ignite/impl/cache/query/query_batch.h \
- ignite/impl/cache/query/query_fields_row_impl.h \
- ignite/impl/cache/query/query_impl.h \
- ignite/impl/cluster/cluster_group_impl.h \
- ignite/impl/handle_registry.h \
- ignite/impl/ignite_binding_impl.h \
- ignite/impl/ignite_environment.h \
- ignite/impl/ignite_impl.h \
- ignite/impl/interop/interop_external_memory.h \
- ignite/impl/interop/interop_target.h \
- ignite/impl/module_manager.h \
- ignite/impl/operations.h \
- ignite/impl/transactions/transactions_impl.h \
- ignite/impl/transactions/transaction_impl.h \
- ignite/transactions/transaction.h \
- ignite/transactions/transactions.h \
- ignite/transactions/transaction_consts.h \
- ignite/transactions/transaction_metrics.h
+ ignite/cache/cache.h \
+ ignite/cache/cache_entry.h \
+ ignite/cache/cache_entry_processor.h \
+ ignite/cache/cache_peek_mode.h \
+ ignite/cache/event/cache_entry_event.h \
+ ignite/cache/event/cache_entry_event_filter.h \
+ ignite/cache/event/cache_entry_event_listener.h \
+ ignite/cache/mutable_cache_entry.h \
+ ignite/cache/query/continuous/continuous_query.h \
+ ignite/cache/query/continuous/continuous_query_handle.h \
+ ignite/cache/query/query.h \
+ ignite/cache/query/query_cursor.h \
+ ignite/cache/query/query_fields_cursor.h \
+ ignite/cache/query/query_fields_row.h \
+ ignite/cache/query/query_scan.h \
+ ignite/cache/query/query_sql.h \
+ ignite/cache/query/query_sql_fields.h \
+ ignite/cache/query/query_text.h \
+ ignite/compute/compute.h \
+ ignite/compute/compute_func.h \
+ ignite/ignite.h \
+ ignite/ignite_binding.h \
+ ignite/ignite_binding_context.h \
+ ignite/ignite_configuration.h \
+ ignite/ignition.h \
+ ignite/impl/binary/binary_type_updater_impl.h \
+ ignite/impl/bindings.h \
+ ignite/impl/cache/cache_entry_processor_holder.h \
+ ignite/impl/cache/cache_impl.h \
+ ignite/impl/cache/event/cache_entry_event_filter_base.h \
+ ignite/impl/cache/event/cache_entry_event_filter_holder.h \
+ ignite/impl/cache/query/continuous/continuous_query_handle_impl.h \
+ ignite/impl/cache/query/continuous/continuous_query_impl.h \
+ ignite/impl/cache/query/query_argument.h \
+ ignite/impl/cache/query/query_batch.h \
+ ignite/impl/cache/query/query_fields_row_impl.h \
+ ignite/impl/cache/query/query_impl.h \
+ ignite/impl/cluster/cluster_group_impl.h \
+ ignite/impl/compute/cancelable_impl.h \
+ ignite/impl/compute/compute_impl.h \
+ ignite/impl/compute/compute_job_holder.h \
+ ignite/impl/compute/compute_job_result.h \
+ ignite/impl/compute/compute_task_holder.h \
+ ignite/impl/handle_registry.h \
+ ignite/impl/ignite_binding_impl.h \
+ ignite/impl/ignite_environment.h \
+ ignite/impl/ignite_impl.h \
+ ignite/impl/interop/interop_external_memory.h \
+ ignite/impl/interop/interop_target.h \
+ ignite/impl/module_manager.h \
+ ignite/impl/operations.h \
+ ignite/impl/transactions/transactions_impl.h \
+ ignite/impl/transactions/transaction_impl.h \
+ ignite/transactions/transaction.h \
+ ignite/transactions/transactions.h \
+ ignite/transactions/transaction_consts.h \
+ ignite/transactions/transaction_metrics.h
uninstall-hook:
if [ -d ${includedir}/ignite ]; then find ${includedir}/ignite -type d -empty -delete; fi
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/cache/cache.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/cache/cache.h b/modules/platforms/cpp/core/include/ignite/cache/cache.h
index e37e806..b16d7f5 100644
--- a/modules/platforms/cpp/core/include/ignite/cache/cache.h
+++ b/modules/platforms/cpp/core/include/ignite/cache/cache.h
@@ -1529,7 +1529,7 @@ namespace ignite
* @code{.cpp}
* IGNITE_EXPORTED_CALL void IgniteModuleInit(ignite::IgniteBindingContext& context)
* {
- * IgniteBinding binding = context.GetBingding();
+ * IgniteBinding binding = context.GetBinding();
*
* binding.RegisterCacheEntryProcessor<MyProcessor1>();
* binding.RegisterCacheEntryProcessor<MyProcessor2>();
@@ -1582,7 +1582,7 @@ namespace ignite
* @code{.cpp}
* IGNITE_EXPORTED_CALL void IgniteModuleInit(ignite::IgniteBindingContext& context)
* {
- * IgniteBinding binding = context.GetBingding();
+ * IgniteBinding binding = context.GetBinding();
*
* binding.RegisterCacheEntryProcessor<MyProcessor1>();
* binding.RegisterCacheEntryProcessor<MyProcessor2>();
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/cache/query/query.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/cache/query/query.h b/modules/platforms/cpp/core/include/ignite/cache/query/query.h
index b2991e4..a5a5c1f 100644
--- a/modules/platforms/cpp/core/include/ignite/cache/query/query.h
+++ b/modules/platforms/cpp/core/include/ignite/cache/query/query.h
@@ -23,7 +23,6 @@
#ifndef _IGNITE_CACHE_QUERY_QUERY
#define _IGNITE_CACHE_QUERY_QUERY
-#include "ignite/cache/query/query_argument.h"
#include "ignite/cache/query/query_cursor.h"
#include "ignite/cache/query/query_scan.h"
#include "ignite/cache/query/query_sql.h"
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/compute/compute.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/compute/compute.h b/modules/platforms/cpp/core/include/ignite/compute/compute.h
new file mode 100644
index 0000000..b079569
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/compute/compute.h
@@ -0,0 +1,136 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::compute::Compute class.
+ */
+
+#ifndef _IGNITE_COMPUTE_COMPUTE
+#define _IGNITE_COMPUTE_COMPUTE
+
+#include <ignite/common/common.h>
+
+#include <ignite/ignite_error.h>
+#include <ignite/future.h>
+#include <ignite/compute/compute_func.h>
+
+#include <ignite/impl/compute/compute_impl.h>
+
+namespace ignite
+{
+ namespace compute
+ {
+ /**
+ * Defines compute grid functionality for executing tasks and closures
+ * over nodes in the ClusterGroup. Instance of Compute is obtained from
+ * Ignite as follows:
+ * @code{.cpp}
+ * Ignite node = Ignition::Get();
+ *
+ * // Compute over all nodes in the cluster.
+ * Compute c = node.GetCompute();
+ * @endcode
+ *
+ * @par Load Balancing
+ * In all cases other than <tt>Broadcast(...)</tt>, Ignite must select
+ * a node for a computation to be executed. The node will be selected
+ * based on the underlying \c LoadBalancingSpi, which by default
+ * sequentially picks next available node from the underlying cluster
+ * group. Other load balancing policies, such as \c random or
+ * \c adaptive, can be configured as well by selecting a different
+ * load balancing SPI in Ignite configuration.
+ *
+ * @par Fault Tolerance
+ * Ignite guarantees that as long as there is at least one grid node
+ * standing, every job will be executed. Jobs will automatically
+ * failover to another node if a remote node crashed or has rejected
+ * execution due to lack of resources. By default, in case of failover,
+ * next load balanced node will be picked for job execution. Also jobs
+ * will never be re-routed to the nodes they have failed on. This
+ * behavior can be changed by configuring any of the existing or a
+ * custom FailoverSpi in grid configuration.
+ *
+ * @par Computation SPIs
+ * Note that regardless of which method is used for executing
+ * computations, all relevant SPI implementations configured for this
+ * compute instance will be used (i.e. failover, load balancing,
+ * collision resolution, checkpoints, etc.).
+ */
+ class IGNITE_IMPORT_EXPORT Compute
+ {
+ public:
+ /**
+ * Constructor.
+ *
+ * Internal method. Should not be used by user.
+ *
+ * @param impl Implementation.
+ */
+ Compute(common::concurrent::SharedPointer<impl::compute::ComputeImpl> impl) :
+ impl(impl)
+ {
+ // No-op.
+ }
+
+ /**
+ * Calls provided ComputeFunc on a node within the underlying
+ * cluster group.
+ *
+ * @tparam R Call return type. BinaryType should be specialized for
+ * the type if it is not primitive. Should not be void. For
+ * non-returning methods see Compute::Run().
+ * @tparam F Compute function type. Should implement ComputeFunc
+ * class.
+ * @param func Compute function to call.
+ * @return Computation result.
+ * @throw IgniteError in case of error.
+ */
+ template<typename R, typename F>
+ R Call(const F& func)
+ {
+ return impl.Get()->CallAsync<R, F>(func).GetValue();
+ }
+
+ /**
+ * Asyncronuously calls provided ComputeFunc on a node within
+ * the underlying cluster group.
+ *
+ * @tparam R Call return type. BinaryType should be specialized for
+ * the type if it is not primitive. Should not be void. For
+ * non-returning methods see Compute::Run().
+ * @tparam F Compute function type. Should implement ComputeFunc
+ * class.
+ * @param func Compute function to call.
+ * @return Future that can be used to access computation result once
+ * it's ready.
+ * @throw IgniteError in case of error.
+ */
+ template<typename R, typename F>
+ Future<R> CallAsync(const F& func)
+ {
+ return impl.Get()->CallAsync<R, F>(func);
+ }
+
+ private:
+ /** Implementation. */
+ common::concurrent::SharedPointer<impl::compute::ComputeImpl> impl;
+ };
+ }
+}
+
+#endif //_IGNITE_COMPUTE_COMPUTE
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/compute/compute_func.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/compute/compute_func.h b/modules/platforms/cpp/core/include/ignite/compute/compute_func.h
new file mode 100644
index 0000000..93a7b4f
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/compute/compute_func.h
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::compute::ComputeFunc class template.
+ */
+
+#ifndef _IGNITE_COMPUTE_COMPUTE_FUNC
+#define _IGNITE_COMPUTE_COMPUTE_FUNC
+
+namespace ignite
+{
+ class IgniteBinding;
+
+ namespace compute
+ {
+ /**
+ * Interface for a simple compute function that can be serialized and
+ * called on the remote nodes. ignite::binary::BinaryType class template
+ * should be specialized for any class, inheriting from this class.
+ *
+ * @tparam R Call return type. BinaryType should be specialized for the
+ * type if it is not primitive.
+ */
+ template<typename R>
+ class ComputeFunc
+ {
+ friend class ignite::IgniteBinding;
+
+ typedef R ReturnType;
+ public:
+ /**
+ * Destructor.
+ */
+ virtual ~ComputeFunc()
+ {
+ // No-op.
+ }
+
+ /**
+ * Called upon execution by compute.
+ *
+ * @return Computation result.
+ */
+ virtual R Call() = 0;
+ };
+ }
+}
+
+#endif //_IGNITE_COMPUTE_COMPUTE_FUNC
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/ignite.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/ignite.h b/modules/platforms/cpp/core/include/ignite/ignite.h
index d96f81f..b3b06f0 100644
--- a/modules/platforms/cpp/core/include/ignite/ignite.h
+++ b/modules/platforms/cpp/core/include/ignite/ignite.h
@@ -23,10 +23,12 @@
#ifndef _IGNITE_IGNITE
#define _IGNITE_IGNITE
-#include "ignite/cache/cache.h"
-#include "ignite/transactions/transactions.h"
-#include "ignite/impl/ignite_impl.h"
-#include "ignite/ignite_configuration.h"
+#include <ignite/impl/ignite_impl.h>
+
+#include <ignite/ignite_configuration.h>
+#include <ignite/cache/cache.h>
+#include <ignite/transactions/transactions.h>
+#include <ignite/compute/compute.h>
namespace ignite
{
@@ -189,6 +191,15 @@ namespace ignite
transactions::Transactions GetTransactions();
/**
+ * Get compute.
+ *
+ * This method should only be called on the valid instance.
+ *
+ * @return Compute class instance.
+ */
+ compute::Compute GetCompute();
+
+ /**
* Get ignite binding.
*
* This method should only be used on the valid instance.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/ignite_binding.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/ignite_binding.h b/modules/platforms/cpp/core/include/ignite/ignite_binding.h
index 344bdf6..fe060c1 100644
--- a/modules/platforms/cpp/core/include/ignite/ignite_binding.h
+++ b/modules/platforms/cpp/core/include/ignite/ignite_binding.h
@@ -121,6 +121,31 @@ namespace ignite
}
/**
+ * Register type as Compute function.
+ *
+ * Registred type should be a child of ignite::compute::ComputeFunc
+ * class.
+ */
+ template<typename F>
+ void RegisterComputeFunc()
+ {
+ impl::IgniteBindingImpl *im = impl.Get();
+
+ int32_t typeId = binary::BinaryType<F>::GetTypeId();
+
+ if (im)
+ {
+ im->RegisterCallback(impl::IgniteBindingImpl::CallbackType::COMPUTE_JOB_CREATE,
+ typeId, impl::binding::ComputeJobCreate<F, typename F::ReturnType>);
+ }
+ else
+ {
+ throw IgniteError(IgniteError::IGNITE_ERR_GENERIC,
+ "Instance is not usable (did you check for error?).");
+ }
+ }
+
+ /**
* Check if the instance is valid.
*
* Invalid instance can be returned if some of the previous operations
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/ignite_binding_context.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/ignite_binding_context.h b/modules/platforms/cpp/core/include/ignite/ignite_binding_context.h
index 4d8a7a7..244b255 100644
--- a/modules/platforms/cpp/core/include/ignite/ignite_binding_context.h
+++ b/modules/platforms/cpp/core/include/ignite/ignite_binding_context.h
@@ -48,7 +48,7 @@ namespace ignite
*
* @return IgniteBinding instance.
*/
- IgniteBinding GetBingding() const
+ IgniteBinding GetBinding() const
{
return binding;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/bindings.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/bindings.h b/modules/platforms/cpp/core/include/ignite/impl/bindings.h
index c1ada12..357b0f0 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/bindings.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/bindings.h
@@ -24,6 +24,7 @@
#include <ignite/impl/ignite_environment.h>
#include <ignite/impl/cache/query/continuous/continuous_query_impl.h>
#include <ignite/impl/cache/cache_entry_processor_holder.h>
+#include <ignite/impl/compute/compute_task_holder.h>
namespace ignite
{
@@ -88,6 +89,29 @@ namespace ignite
return 0;
}
+
+ /**
+ * Binding for compute job creation.
+ *
+ * @tparam F The job type.
+ * @tparam R The job return type.
+ *
+ * @param reader Reader.
+ * @param env Environment.
+ * @return Handle for the job.
+ */
+ template<typename F, typename R>
+ int64_t ComputeJobCreate(binary::BinaryReaderImpl& reader, binary::BinaryWriterImpl&, IgniteEnvironment& env)
+ {
+ using namespace common::concurrent;
+ using namespace compute;
+
+ F job = reader.ReadObject<F>();
+
+ SharedPointer<ComputeJobHolder> jobPtr(new ComputeJobHolderImpl<F, R>(job));
+
+ return env.GetHandleRegistry().Allocate(jobPtr);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/cluster/cluster_group_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/cluster/cluster_group_impl.h b/modules/platforms/cpp/core/include/ignite/impl/cluster/cluster_group_impl.h
index 2bff0d8..3cfd700 100644
--- a/modules/platforms/cpp/core/include/ignite/impl/cluster/cluster_group_impl.h
+++ b/modules/platforms/cpp/core/include/ignite/impl/cluster/cluster_group_impl.h
@@ -22,6 +22,7 @@
#include <ignite/jni/java.h>
#include <ignite/impl/interop/interop_target.h>
+#include <ignite/impl/compute/compute_impl.h>
namespace ignite
{
@@ -29,13 +30,19 @@ namespace ignite
{
namespace cluster
{
+ /* Forward declaration. */
+ class ClusterGroupImpl;
+
+ /* Shared pointer. */
+ typedef common::concurrent::SharedPointer<ClusterGroupImpl> SP_ClusterGroupImpl;
+
/**
* Cluster group implementation.
*/
class IGNITE_FRIEND_EXPORT ClusterGroupImpl : private interop::InteropTarget
{
typedef common::concurrent::SharedPointer<IgniteEnvironment> SP_IgniteEnvironment;
- typedef common::concurrent::SharedPointer<ClusterGroupImpl> SP_ClusterGroupImpl;
+ typedef common::concurrent::SharedPointer<compute::ComputeImpl> SP_ComputeImpl;
public:
/**
* Constructor used to create new instance.
@@ -53,12 +60,20 @@ namespace ignite
/**
* Get server nodes cluster group implementation.
*
- * @param err Error.
* @return Server nodes cluster group implementation.
*/
- SP_ClusterGroupImpl ForServers(IgniteError& err);
+ SP_ClusterGroupImpl ForServers();
+
+ /**
+ * Get compute instance over this cluster group.
+ *
+ * @return Compute instance.
+ */
+ SP_ComputeImpl GetCompute();
private:
+ IGNITE_NO_COPY_ASSIGNMENT(ClusterGroupImpl);
+
/**
* Make cluster group implementation using java reference and
* internal state of this cluster group.
@@ -68,7 +83,15 @@ namespace ignite
*/
SP_ClusterGroupImpl FromTarget(jobject javaRef);
- IGNITE_NO_COPY_ASSIGNMENT(ClusterGroupImpl)
+ /**
+ * Gets instance of compute internally.
+ *
+ * @return Instance of compute.
+ */
+ SP_ComputeImpl InternalGetCompute();
+
+ /** Compute for the cluster group. */
+ SP_ComputeImpl computeImpl;
};
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/compute/cancelable_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/compute/cancelable_impl.h b/modules/platforms/cpp/core/include/ignite/impl/compute/cancelable_impl.h
new file mode 100644
index 0000000..828ecba
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/impl/compute/cancelable_impl.h
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::impl::compute::CancelableImpl class.
+ */
+
+#ifndef _IGNITE_IMPL_COMPUTE_CANCELABLE_IMPL
+#define _IGNITE_IMPL_COMPUTE_CANCELABLE_IMPL
+
+#include <ignite/common/common.h>
+#include <ignite/common/cancelable.h>
+#include <ignite/impl/interop/interop_target.h>
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ /**
+ * Compute implementation.
+ */
+ class IGNITE_IMPORT_EXPORT CancelableImpl : public interop::InteropTarget, public common::Cancelable
+ {
+ public:
+ /**
+ * Constructor.
+ *
+ * @param env Environment.
+ * @param javaRef Java object reference.
+ */
+ CancelableImpl(common::concurrent::SharedPointer<IgniteEnvironment> env, jobject javaRef);
+
+ /**
+ * Destructor.
+ */
+ virtual ~CancelableImpl()
+ {
+ // No-op.
+ }
+
+ /**
+ * Cancels the operation.
+ */
+ virtual void Cancel();
+
+ private:
+ IGNITE_NO_COPY_ASSIGNMENT(CancelableImpl);
+ };
+ }
+ }
+}
+
+#endif //_IGNITE_IMPL_COMPUTE_CANCELABLE_IMPL
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9c96de5/modules/platforms/cpp/core/include/ignite/impl/compute/compute_impl.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/core/include/ignite/impl/compute/compute_impl.h b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_impl.h
new file mode 100644
index 0000000..389c571
--- /dev/null
+++ b/modules/platforms/cpp/core/include/ignite/impl/compute/compute_impl.h
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+/**
+ * @file
+ * Declares ignite::impl::compute::ComputeImpl class.
+ */
+
+#ifndef _IGNITE_IMPL_COMPUTE_COMPUTE_IMPL
+#define _IGNITE_IMPL_COMPUTE_COMPUTE_IMPL
+
+#include <ignite/common/common.h>
+#include <ignite/common/promise.h>
+#include <ignite/impl/interop/interop_target.h>
+#include <ignite/impl/compute/compute_task_holder.h>
+#include <ignite/impl/compute/cancelable_impl.h>
+
+#include <ignite/ignite_error.h>
+
+namespace ignite
+{
+ namespace impl
+ {
+ namespace compute
+ {
+ /**
+ * Compute implementation.
+ */
+ class IGNITE_IMPORT_EXPORT ComputeImpl : public interop::InteropTarget
+ {
+ public:
+ /**
+ * Operation type.
+ */
+ struct Operation
+ {
+ enum Type
+ {
+ Unicast = 5
+ };
+ };
+
+ /**
+ * Constructor.
+ *
+ * @param env Environment.
+ * @param javaRef Java object reference.
+ */
+ ComputeImpl(common::concurrent::SharedPointer<IgniteEnvironment> env, jobject javaRef);
+
+ /**
+ * Asyncronuously calls provided ComputeFunc on a node within
+ * the underlying cluster group.
+ *
+ * @tparam F Compute function type. Should implement ComputeFunc
+ * class.
+ * @tparam R Call return type. BinaryType should be specialized for
+ * the type if it is not primitive. Should not be void. For
+ * non-returning methods see Compute::Run().
+ * @param func Compute function to call.
+ * @return Future that can be used to acess computation result once
+ * it's ready.
+ * @throw IgniteError in case of error.
+ */
+ template<typename R, typename F>
+ Future<R> CallAsync(const F& func)
+ {
+ common::concurrent::SharedPointer<interop::InteropMemory> mem = GetEnvironment().AllocateMemory();
+ interop::InteropOutputStream out(mem.Get());
+ binary::BinaryWriterImpl writer(&out, GetEnvironment().GetTypeManager());
+
+ common::concurrent::SharedPointer<ComputeJobHolder> job(new ComputeJobHolderImpl<F, R>(func));
+
+ int64_t jobHandle = GetEnvironment().GetHandleRegistry().Allocate(job);
+
+ ComputeTaskHolderImpl<F, R>* taskPtr = new ComputeTaskHolderImpl<F, R>(jobHandle);
+ common::concurrent::SharedPointer<ComputeTaskHolder> task(taskPtr);
+
+ int64_t taskHandle = GetEnvironment().GetHandleRegistry().Allocate(task);
+
+ writer.WriteInt64(taskHandle);
+ writer.WriteInt32(1);
+ writer.WriteInt64(jobHandle);
+ writer.WriteObject<F>(func);
+
+ out.Synchronize();
+
+ jobject target = InStreamOutObject(Operation::Unicast, *mem.Get());
+ std::auto_ptr<common::Cancelable> cancelable(new CancelableImpl(GetEnvironmentPointer(), target));
+
+ common::Promise<R>& promise = taskPtr->GetPromise();
+ promise.SetCancelTarget(cancelable);
+
+ return promise.GetFuture();
+ }
+
+ private:
+ IGNITE_NO_COPY_ASSIGNMENT(ComputeImpl);
+ };
+ }
+ }
+}
+
+#endif //_IGNITE_IMPL_COMPUTE_COMPUTE_IMPL