You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2016/12/12 14:45:06 UTC
[1/3] ignite git commit: IGNITE-4033 Streamline platform callback
interface
Repository: ignite
Updated Branches:
refs/heads/master f621f7f76 -> 72ac53da2
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbacks.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbacks.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbacks.cs
index cc205e8..09933be 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbacks.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbacks.cs
@@ -22,7 +22,6 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
- using System.IO;
using System.Runtime.InteropServices;
using System.Threading;
using Apache.Ignite.Core.Cache.Affinity;
@@ -77,7 +76,14 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
/** Keep references to created delegates. */
// ReSharper disable once CollectionNeverQueried.Local
- private readonly List<Delegate> _delegates = new List<Delegate>(50);
+ private readonly List<Delegate> _delegates = new List<Delegate>(5);
+
+ /** Handlers array. */
+ private readonly InLongOutLongHandler[] _inLongOutLongHandlers = new InLongOutLongHandler[62];
+
+ /** Handlers array. */
+ private readonly InLongLongLongObjectOutLongHandler[] _inLongLongLongObjectOutLongHandlers
+ = new InLongLongLongObjectOutLongHandler[62];
/** Initialized flag. */
private readonly ManualResetEventSlim _initEvent = new ManualResetEventSlim(false);
@@ -107,90 +113,19 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
/** Operation: prepare .Net. */
private const int OpPrepareDotNet = 1;
- private delegate long CacheStoreCreateCallbackDelegate(void* target, long memPtr);
- private delegate int CacheStoreInvokeCallbackDelegate(void* target, long objPtr, long memPtr);
- private delegate void CacheStoreDestroyCallbackDelegate(void* target, long objPtr);
- private delegate long CacheStoreSessionCreateCallbackDelegate(void* target, long storePtr);
-
- private delegate long CacheEntryFilterCreateCallbackDelegate(void* target, long memPtr);
- private delegate int CacheEntryFilterApplyCallbackDelegate(void* target, long objPtr, long memPtr);
- private delegate void CacheEntryFilterDestroyCallbackDelegate(void* target, long objPtr);
-
- private delegate void CacheInvokeCallbackDelegate(void* target, long inMemPtr, long outMemPtr);
-
- private delegate void ComputeTaskMapCallbackDelegate(void* target, long taskPtr, long inMemPtr, long outMemPtr);
- private delegate int ComputeTaskJobResultCallbackDelegate(void* target, long taskPtr, long jobPtr, long memPtr);
- private delegate void ComputeTaskReduceCallbackDelegate(void* target, long taskPtr);
- private delegate void ComputeTaskCompleteCallbackDelegate(void* target, long taskPtr, long memPtr);
- private delegate int ComputeJobSerializeCallbackDelegate(void* target, long jobPtr, long memPtr);
- private delegate long ComputeJobCreateCallbackDelegate(void* target, long memPtr);
- private delegate void ComputeJobExecuteCallbackDelegate(void* target, long jobPtr, int cancel, long memPtr);
- private delegate void ComputeJobCancelCallbackDelegate(void* target, long jobPtr);
- private delegate void ComputeJobDestroyCallbackDelegate(void* target, long jobPtr);
-
- private delegate void ContinuousQueryListenerApplyCallbackDelegate(void* target, long lsnrPtr, long memPtr);
- private delegate long ContinuousQueryFilterCreateCallbackDelegate(void* target, long memPtr);
- private delegate int ContinuousQueryFilterApplyCallbackDelegate(void* target, long filterPtr, long memPtr);
- private delegate void ContinuousQueryFilterReleaseCallbackDelegate(void* target, long filterPtr);
-
- private delegate void DataStreamerTopologyUpdateCallbackDelegate(void* target, long ldrPtr, long topVer, int topSize);
- private delegate void DataStreamerStreamReceiverInvokeCallbackDelegate(void* target, long ptr, void* cache, long memPtr, byte keepPortable);
-
- private delegate void FutureByteResultCallbackDelegate(void* target, long futPtr, int res);
- private delegate void FutureBoolResultCallbackDelegate(void* target, long futPtr, int res);
- private delegate void FutureShortResultCallbackDelegate(void* target, long futPtr, int res);
- private delegate void FutureCharResultCallbackDelegate(void* target, long futPtr, int res);
- private delegate void FutureIntResultCallbackDelegate(void* target, long futPtr, int res);
- private delegate void FutureFloatResultCallbackDelegate(void* target, long futPtr, float res);
- private delegate void FutureLongResultCallbackDelegate(void* target, long futPtr, long res);
- private delegate void FutureDoubleResultCallbackDelegate(void* target, long futPtr, double res);
- private delegate void FutureObjectResultCallbackDelegate(void* target, long futPtr, long memPtr);
- private delegate void FutureNullResultCallbackDelegate(void* target, long futPtr);
- private delegate void FutureErrorCallbackDelegate(void* target, long futPtr, long memPtr);
-
- private delegate void LifecycleOnEventCallbackDelegate(void* target, long ptr, int evt);
-
- private delegate void MemoryReallocateCallbackDelegate(void* target, long memPtr, int cap);
-
- private delegate long MessagingFilterCreateCallbackDelegate(void* target, long memPtr);
- private delegate int MessagingFilterApplyCallbackDelegate(void* target, long ptr, long memPtr);
- private delegate void MessagingFilterDestroyCallbackDelegate(void* target, long ptr);
-
- private delegate long EventFilterCreateCallbackDelegate(void* target, long memPtr);
- private delegate int EventFilterApplyCallbackDelegate(void* target, long ptr, long memPtr);
- private delegate void EventFilterDestroyCallbackDelegate(void* target, long ptr);
-
- private delegate long ServiceInitCallbackDelegate(void* target, long memPtr);
- private delegate void ServiceExecuteCallbackDelegate(void* target, long svcPtr, long memPtr);
- private delegate void ServiceCancelCallbackDelegate(void* target, long svcPtr, long memPtr);
- private delegate void ServiceInvokeMethodCallbackDelegate(void* target, long svcPtr, long inMemPtr, long outMemPtr);
-
- private delegate int ClusterNodeFilterApplyCallbackDelegate(void* target, long memPtr);
-
- private delegate void NodeInfoCallbackDelegate(void* target, long memPtr);
-
- private delegate void OnStartCallbackDelegate(void* target, void* proc, long memPtr);
- private delegate void OnStopCallbackDelegate(void* target);
-
private delegate void ErrorCallbackDelegate(void* target, int errType, sbyte* errClsChars, int errClsCharsLen, sbyte* errMsgChars, int errMsgCharsLen, sbyte* stackTraceChars, int stackTraceCharsLen, void* errData, int errDataLen);
- private delegate long ExtensionCallbackInLongOutLongDelegate(void* target, int typ, long arg1);
- private delegate long ExtensionCallbackInLongLongOutLongDelegate(void* target, int typ, long arg1, long arg2);
-
- private delegate void OnClientDisconnectedDelegate(void* target);
- private delegate void OnClientReconnectedDelegate(void* target, bool clusterRestarted);
-
private delegate void LoggerLogDelegate(void* target, int level, sbyte* messageChars, int messageCharsLen, sbyte* categoryChars, int categoryCharsLen, sbyte* errorInfoChars, int errorInfoCharsLen, long memPtr);
private delegate bool LoggerIsLevelEnabledDelegate(void* target, int level);
- private delegate long AffinityFunctionInitDelegate(void* target, long memPtr, void* baseFunc);
- private delegate int AffinityFunctionPartitionDelegate(void* target, long ptr, long memPtr);
- private delegate void AffinityFunctionAssignPartitionsDelegate(void* target, long ptr, long inMemPtr, long outMemPtr);
- private delegate void AffinityFunctionRemoveNodeDelegate(void* target, long ptr, long memPtr);
- private delegate void AffinityFunctionDestroyDelegate(void* target, long ptr);
-
private delegate void ConsoleWriteDelegate(sbyte* chars, int charsLen, bool isErr);
+ private delegate long InLongOutLongDelegate(void* target, int type, long val);
+ private delegate long InLongLongLongObjectOutLongDelegate(void* target, int type, long val1, long val2, long val3, void* arg);
+
+ private delegate long InLongOutLongFunc(long val);
+ private delegate long InLongLongLongObjectOutLongFunc(long val1, long val2, long val3, void* arg);
+
/// <summary>
/// Constructor.
/// </summary>
@@ -204,89 +139,13 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
{
target = IntPtr.Zero.ToPointer(), // Target is not used in .Net as we rely on dynamic FP creation.
- cacheStoreCreate = CreateFunctionPointer((CacheStoreCreateCallbackDelegate) CacheStoreCreate),
- cacheStoreInvoke = CreateFunctionPointer((CacheStoreInvokeCallbackDelegate) CacheStoreInvoke),
- cacheStoreDestroy = CreateFunctionPointer((CacheStoreDestroyCallbackDelegate) CacheStoreDestroy),
-
- cacheStoreSessionCreate = CreateFunctionPointer((CacheStoreSessionCreateCallbackDelegate) CacheStoreSessionCreate),
-
- cacheEntryFilterCreate = CreateFunctionPointer((CacheEntryFilterCreateCallbackDelegate)CacheEntryFilterCreate),
- cacheEntryFilterApply = CreateFunctionPointer((CacheEntryFilterApplyCallbackDelegate)CacheEntryFilterApply),
- cacheEntryFilterDestroy = CreateFunctionPointer((CacheEntryFilterDestroyCallbackDelegate)CacheEntryFilterDestroy),
-
- cacheInvoke = CreateFunctionPointer((CacheInvokeCallbackDelegate) CacheInvoke),
-
- computeTaskMap = CreateFunctionPointer((ComputeTaskMapCallbackDelegate) ComputeTaskMap),
- computeTaskJobResult =
- CreateFunctionPointer((ComputeTaskJobResultCallbackDelegate) ComputeTaskJobResult),
- computeTaskReduce = CreateFunctionPointer((ComputeTaskReduceCallbackDelegate) ComputeTaskReduce),
- computeTaskComplete = CreateFunctionPointer((ComputeTaskCompleteCallbackDelegate) ComputeTaskComplete),
- computeJobSerialize = CreateFunctionPointer((ComputeJobSerializeCallbackDelegate) ComputeJobSerialize),
- computeJobCreate = CreateFunctionPointer((ComputeJobCreateCallbackDelegate) ComputeJobCreate),
- computeJobExecute = CreateFunctionPointer((ComputeJobExecuteCallbackDelegate) ComputeJobExecute),
- computeJobCancel = CreateFunctionPointer((ComputeJobCancelCallbackDelegate) ComputeJobCancel),
- computeJobDestroy = CreateFunctionPointer((ComputeJobDestroyCallbackDelegate) ComputeJobDestroy),
- continuousQueryListenerApply =
- CreateFunctionPointer((ContinuousQueryListenerApplyCallbackDelegate) ContinuousQueryListenerApply),
- continuousQueryFilterCreate =
- CreateFunctionPointer((ContinuousQueryFilterCreateCallbackDelegate) ContinuousQueryFilterCreate),
- continuousQueryFilterApply =
- CreateFunctionPointer((ContinuousQueryFilterApplyCallbackDelegate) ContinuousQueryFilterApply),
- continuousQueryFilterRelease =
- CreateFunctionPointer((ContinuousQueryFilterReleaseCallbackDelegate) ContinuousQueryFilterRelease),
- dataStreamerTopologyUpdate =
- CreateFunctionPointer((DataStreamerTopologyUpdateCallbackDelegate) DataStreamerTopologyUpdate),
- dataStreamerStreamReceiverInvoke =
- CreateFunctionPointer((DataStreamerStreamReceiverInvokeCallbackDelegate) DataStreamerStreamReceiverInvoke),
-
- futureByteResult = CreateFunctionPointer((FutureByteResultCallbackDelegate) FutureByteResult),
- futureBoolResult = CreateFunctionPointer((FutureBoolResultCallbackDelegate) FutureBoolResult),
- futureShortResult = CreateFunctionPointer((FutureShortResultCallbackDelegate) FutureShortResult),
- futureCharResult = CreateFunctionPointer((FutureCharResultCallbackDelegate) FutureCharResult),
- futureIntResult = CreateFunctionPointer((FutureIntResultCallbackDelegate) FutureIntResult),
- futureFloatResult = CreateFunctionPointer((FutureFloatResultCallbackDelegate) FutureFloatResult),
- futureLongResult = CreateFunctionPointer((FutureLongResultCallbackDelegate) FutureLongResult),
- futureDoubleResult = CreateFunctionPointer((FutureDoubleResultCallbackDelegate) FutureDoubleResult),
- futureObjectResult = CreateFunctionPointer((FutureObjectResultCallbackDelegate) FutureObjectResult),
- futureNullResult = CreateFunctionPointer((FutureNullResultCallbackDelegate) FutureNullResult),
- futureError = CreateFunctionPointer((FutureErrorCallbackDelegate) FutureError),
- lifecycleOnEvent = CreateFunctionPointer((LifecycleOnEventCallbackDelegate) LifecycleOnEvent),
- memoryReallocate = CreateFunctionPointer((MemoryReallocateCallbackDelegate) MemoryReallocate),
- nodeInfo = CreateFunctionPointer((NodeInfoCallbackDelegate) NodeInfo),
-
- messagingFilterCreate = CreateFunctionPointer((MessagingFilterCreateCallbackDelegate)MessagingFilterCreate),
- messagingFilterApply = CreateFunctionPointer((MessagingFilterApplyCallbackDelegate)MessagingFilterApply),
- messagingFilterDestroy = CreateFunctionPointer((MessagingFilterDestroyCallbackDelegate)MessagingFilterDestroy),
-
- eventFilterCreate = CreateFunctionPointer((EventFilterCreateCallbackDelegate)EventFilterCreate),
- eventFilterApply = CreateFunctionPointer((EventFilterApplyCallbackDelegate)EventFilterApply),
- eventFilterDestroy = CreateFunctionPointer((EventFilterDestroyCallbackDelegate)EventFilterDestroy),
-
- serviceInit = CreateFunctionPointer((ServiceInitCallbackDelegate)ServiceInit),
- serviceExecute = CreateFunctionPointer((ServiceExecuteCallbackDelegate)ServiceExecute),
- serviceCancel = CreateFunctionPointer((ServiceCancelCallbackDelegate)ServiceCancel),
- serviceInvokeMethod = CreateFunctionPointer((ServiceInvokeMethodCallbackDelegate)ServiceInvokeMethod),
-
- clusterNodeFilterApply = CreateFunctionPointer((ClusterNodeFilterApplyCallbackDelegate)ClusterNodeFilterApply),
-
- onStart = CreateFunctionPointer((OnStartCallbackDelegate)OnStart),
- onStop = CreateFunctionPointer((OnStopCallbackDelegate)OnStop),
error = CreateFunctionPointer((ErrorCallbackDelegate)Error),
- extensionCbInLongOutLong = CreateFunctionPointer((ExtensionCallbackInLongOutLongDelegate)ExtensionCallbackInLongOutLong),
- extensionCbInLongLongOutLong = CreateFunctionPointer((ExtensionCallbackInLongLongOutLongDelegate)ExtensionCallbackInLongLongOutLong),
-
- onClientDisconnected = CreateFunctionPointer((OnClientDisconnectedDelegate)OnClientDisconnected),
- ocClientReconnected = CreateFunctionPointer((OnClientReconnectedDelegate)OnClientReconnected),
-
- affinityFunctionInit = CreateFunctionPointer((AffinityFunctionInitDelegate)AffinityFunctionInit),
- affinityFunctionPartition = CreateFunctionPointer((AffinityFunctionPartitionDelegate)AffinityFunctionPartition),
- affinityFunctionAssignPartitions = CreateFunctionPointer((AffinityFunctionAssignPartitionsDelegate)AffinityFunctionAssignPartitions),
- affinityFunctionRemoveNode = CreateFunctionPointer((AffinityFunctionRemoveNodeDelegate)AffinityFunctionRemoveNode),
- affinityFunctionDestroy = CreateFunctionPointer((AffinityFunctionDestroyDelegate)AffinityFunctionDestroy),
-
loggerLog = CreateFunctionPointer((LoggerLogDelegate)LoggerLog),
- loggerIsLevelEnabled = CreateFunctionPointer((LoggerIsLevelEnabledDelegate)LoggerIsLevelEnabled)
+ loggerIsLevelEnabled = CreateFunctionPointer((LoggerIsLevelEnabledDelegate)LoggerIsLevelEnabled),
+
+ inLongOutLong = CreateFunctionPointer((InLongOutLongDelegate)InLongOutLong),
+ inLongLongObjectOutLong = CreateFunctionPointer((InLongLongLongObjectOutLongDelegate)InLongLongLongObjectOutLong)
};
_cbsPtr = Marshal.AllocHGlobal(UU.HandlersSize());
@@ -294,6 +153,8 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
Marshal.StructureToPtr(cbs, _cbsPtr, false);
_thisHnd = GCHandle.Alloc(this);
+
+ InitHandlers();
}
/// <summary>
@@ -304,106 +165,257 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
get { return _handleRegistry; }
}
- #region IMPLEMENTATION: CACHE
+ #region HANDLERS
+
+ /// <summary>
+ /// Initializes the handlers.
+ /// </summary>
+ private void InitHandlers()
+ {
+ AddHandler(UnmanagedCallbackOp.CacheStoreCreate, CacheStoreCreate, true);
+ AddHandler(UnmanagedCallbackOp.CacheStoreInvoke, CacheStoreInvoke);
+ AddHandler(UnmanagedCallbackOp.CacheStoreDestroy, CacheStoreDestroy);
+ AddHandler(UnmanagedCallbackOp.CacheStoreSessionCreate, CacheStoreSessionCreate);
+ AddHandler(UnmanagedCallbackOp.CacheEntryFilterCreate, CacheEntryFilterCreate);
+ AddHandler(UnmanagedCallbackOp.CacheEntryFilterApply, CacheEntryFilterApply);
+ AddHandler(UnmanagedCallbackOp.CacheEntryFilterDestroy, CacheEntryFilterDestroy);
+ AddHandler(UnmanagedCallbackOp.CacheInvoke, CacheInvoke);
+ AddHandler(UnmanagedCallbackOp.ComputeTaskMap, ComputeTaskMap);
+ AddHandler(UnmanagedCallbackOp.ComputeTaskJobResult, ComputeTaskJobResult);
+ AddHandler(UnmanagedCallbackOp.ComputeTaskReduce, ComputeTaskReduce);
+ AddHandler(UnmanagedCallbackOp.ComputeTaskComplete, ComputeTaskComplete);
+ AddHandler(UnmanagedCallbackOp.ComputeJobSerialize, ComputeJobSerialize);
+ AddHandler(UnmanagedCallbackOp.ComputeJobCreate, ComputeJobCreate);
+ AddHandler(UnmanagedCallbackOp.ComputeJobExecute, ComputeJobExecute);
+ AddHandler(UnmanagedCallbackOp.ComputeJobCancel, ComputeJobCancel);
+ AddHandler(UnmanagedCallbackOp.ComputeJobDestroy, ComputeJobDestroy);
+ AddHandler(UnmanagedCallbackOp.ContinuousQueryListenerApply, ContinuousQueryListenerApply);
+ AddHandler(UnmanagedCallbackOp.ContinuousQueryFilterCreate, ContinuousQueryFilterCreate);
+ AddHandler(UnmanagedCallbackOp.ContinuousQueryFilterApply, ContinuousQueryFilterApply);
+ AddHandler(UnmanagedCallbackOp.ContinuousQueryFilterRelease, ContinuousQueryFilterRelease);
+ AddHandler(UnmanagedCallbackOp.DataStreamerTopologyUpdate, DataStreamerTopologyUpdate);
+ AddHandler(UnmanagedCallbackOp.DataStreamerStreamReceiverInvoke, DataStreamerStreamReceiverInvoke);
+ AddHandler(UnmanagedCallbackOp.FutureByteResult, FutureByteResult);
+ AddHandler(UnmanagedCallbackOp.FutureBoolResult, FutureBoolResult);
+ AddHandler(UnmanagedCallbackOp.FutureShortResult, FutureShortResult);
+ AddHandler(UnmanagedCallbackOp.FutureCharResult, FutureCharResult);
+ AddHandler(UnmanagedCallbackOp.FutureIntResult, FutureIntResult);
+ AddHandler(UnmanagedCallbackOp.FutureFloatResult, FutureFloatResult);
+ AddHandler(UnmanagedCallbackOp.FutureLongResult, FutureLongResult);
+ AddHandler(UnmanagedCallbackOp.FutureDoubleResult, FutureDoubleResult);
+ AddHandler(UnmanagedCallbackOp.FutureObjectResult, FutureObjectResult);
+ AddHandler(UnmanagedCallbackOp.FutureNullResult, FutureNullResult);
+ AddHandler(UnmanagedCallbackOp.FutureError, FutureError);
+ AddHandler(UnmanagedCallbackOp.LifecycleOnEvent, LifecycleOnEvent, true);
+ AddHandler(UnmanagedCallbackOp.MemoryReallocate, MemoryReallocate, true);
+ AddHandler(UnmanagedCallbackOp.MessagingFilterCreate, MessagingFilterCreate);
+ AddHandler(UnmanagedCallbackOp.MessagingFilterApply, MessagingFilterApply);
+ AddHandler(UnmanagedCallbackOp.MessagingFilterDestroy, MessagingFilterDestroy);
+ AddHandler(UnmanagedCallbackOp.EventFilterCreate, EventFilterCreate);
+ AddHandler(UnmanagedCallbackOp.EventFilterApply, EventFilterApply);
+ AddHandler(UnmanagedCallbackOp.EventFilterDestroy, EventFilterDestroy);
+ AddHandler(UnmanagedCallbackOp.ServiceInit, ServiceInit);
+ AddHandler(UnmanagedCallbackOp.ServiceExecute, ServiceExecute);
+ AddHandler(UnmanagedCallbackOp.ServiceCancel, ServiceCancel);
+ AddHandler(UnmanagedCallbackOp.ServiceInvokeMethod, ServiceInvokeMethod);
+ AddHandler(UnmanagedCallbackOp.ClusterNodeFilterApply, ClusterNodeFilterApply);
+ AddHandler(UnmanagedCallbackOp.NodeInfo, NodeInfo);
+ AddHandler(UnmanagedCallbackOp.OnStart, OnStart, true);
+ AddHandler(UnmanagedCallbackOp.OnStop, OnStop, true);
+ AddHandler(UnmanagedCallbackOp.ExtensionInLongLongOutLong, ExtensionCallbackInLongLongOutLong, true);
+ AddHandler(UnmanagedCallbackOp.OnClientDisconnected, OnClientDisconnected);
+ AddHandler(UnmanagedCallbackOp.OnClientReconnected, OnClientReconnected);
+ AddHandler(UnmanagedCallbackOp.AffinityFunctionInit, AffinityFunctionInit);
+ AddHandler(UnmanagedCallbackOp.AffinityFunctionPartition, AffinityFunctionPartition);
+ AddHandler(UnmanagedCallbackOp.AffinityFunctionAssignPartitions, AffinityFunctionAssignPartitions);
+ AddHandler(UnmanagedCallbackOp.AffinityFunctionRemoveNode, AffinityFunctionRemoveNode);
+ AddHandler(UnmanagedCallbackOp.AffinityFunctionDestroy, AffinityFunctionDestroy);
+ AddHandler(UnmanagedCallbackOp.ComputeTaskLocalJobResult, ComputeTaskLocalJobResult);
+ AddHandler(UnmanagedCallbackOp.ComputeJobExecuteLocal, ComputeJobExecuteLocal);
+ }
+
+ /// <summary>
+ /// Adds the handler.
+ /// </summary>
+ private void AddHandler(UnmanagedCallbackOp op, InLongOutLongFunc func, bool allowUninitialized = false)
+ {
+ _inLongOutLongHandlers[(int)op] = new InLongOutLongHandler(func, allowUninitialized);
+ }
+
+ /// <summary>
+ /// Adds the handler.
+ /// </summary>
+ private void AddHandler(UnmanagedCallbackOp op, InLongLongLongObjectOutLongFunc func,
+ bool allowUninitialized = false)
+ {
+ _inLongLongLongObjectOutLongHandlers[(int)op]
+ = new InLongLongLongObjectOutLongHandler(func, allowUninitialized);
+ }
+
+ #endregion
+
+ #region IMPLEMENTATION: GENERAL PURPOSE
+
+ [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
+ private long InLongOutLong(void* target, int type, long val)
+ {
+ try
+ {
+ if (type < 0 || type > _inLongOutLongHandlers.Length)
+ throw GetInvalidOpError("InLongOutLong", type);
+
+ var hnd = _inLongOutLongHandlers[type];
+
+ if (hnd.Handler == null)
+ throw GetInvalidOpError("InLongOutLong", type);
- private long CacheStoreCreate(void* target, long memPtr)
+ if (!hnd.AllowUninitialized)
+ _initEvent.Wait();
+
+ return hnd.Handler(val);
+ }
+ catch (Exception e)
+ {
+ _log.Error(e, "Failure in Java callback");
+
+ UU.ThrowToJava(_ctx.NativeContext, e);
+
+ return 0;
+ }
+ }
+
+ [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
+ private long InLongLongLongObjectOutLong(void* target, int type, long val1, long val2, long val3, void* arg)
{
- return SafeCall(() =>
+ try
{
- var cacheStore = CacheStore.CreateInstance(memPtr, _handleRegistry);
+ if (type < 0 || type > _inLongLongLongObjectOutLongHandlers.Length)
+ throw GetInvalidOpError("InLongLongLongObjectOutLong", type);
+
+ var hnd = _inLongLongLongObjectOutLongHandlers[type];
+
+ if (hnd.Handler == null)
+ throw GetInvalidOpError("InLongLongLongObjectOutLong", type);
+
+ if (!hnd.AllowUninitialized)
+ _initEvent.Wait();
+
+ return hnd.Handler(val1, val2, val3, arg);
+ }
+ catch (Exception e)
+ {
+ _log.Error(e, "Failure in Java callback");
+
+ UU.ThrowToJava(_ctx.NativeContext, e);
+
+ return 0;
+ }
+ }
- if (_ignite != null)
- cacheStore.Init(_ignite);
- else
+ /// <summary>
+ /// Throws the invalid op error.
+ /// </summary>
+ private static Exception GetInvalidOpError(string method, int type)
+ {
+ return new InvalidOperationException(
+ string.Format("Invalid {0} callback code: {1}", method, (UnmanagedCallbackOp) type));
+ }
+
+ #endregion
+
+ #region IMPLEMENTATION: CACHE
+
+ private long CacheStoreCreate(long memPtr)
+ {
+ var cacheStore = CacheStore.CreateInstance(memPtr, _handleRegistry);
+
+ if (_ignite != null)
+ cacheStore.Init(_ignite);
+ else
+ {
+ lock (_initActions)
{
- lock (_initActions)
- {
- if (_ignite != null)
- cacheStore.Init(_ignite);
- else
- _initActions.Add(cacheStore.Init);
- }
+ if (_ignite != null)
+ cacheStore.Init(_ignite);
+ else
+ _initActions.Add(cacheStore.Init);
}
+ }
- return cacheStore.Handle;
- }, true);
+ return cacheStore.Handle;
}
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
[SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope")]
- private int CacheStoreInvoke(void* target, long objPtr, long memPtr)
+ private long CacheStoreInvoke(long memPtr)
{
- return SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var t = _handleRegistry.Get<CacheStore>(objPtr, true);
+ try
+ {
+ var store = _handleRegistry.Get<CacheStore>(stream.ReadLong(), true);
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
+ return store.Invoke(stream, _ignite);
+ }
+ catch (Exception e)
{
- try
- {
- return t.Invoke(stream, _ignite);
- }
- catch (Exception e)
- {
- stream.Seek(0, SeekOrigin.Begin);
+ stream.Reset();
- _ignite.Marshaller.StartMarshal(stream).WriteObject(e);
+ _ignite.Marshaller.StartMarshal(stream).WriteObject(e);
- return -1;
- }
+ return -1;
}
- });
+ }
}
- private void CacheStoreDestroy(void* target, long objPtr)
+ private long CacheStoreDestroy(long objPtr)
{
- SafeCall(() => _ignite.HandleRegistry.Release(objPtr));
+ _ignite.HandleRegistry.Release(objPtr);
+
+ return 0;
}
- private long CacheStoreSessionCreate(void* target, long storePtr)
+ private long CacheStoreSessionCreate(long val)
{
- return SafeCall(() => _ignite.HandleRegistry.Allocate(new CacheStoreSession()));
+ return _ignite.HandleRegistry.Allocate(new CacheStoreSession());
}
- private long CacheEntryFilterCreate(void* target, long memPtr)
+ private long CacheEntryFilterCreate(long memPtr)
{
- return SafeCall(() => _handleRegistry.Allocate(CacheEntryFilterHolder.CreateInstance(memPtr, _ignite)));
+ return _handleRegistry.Allocate(CacheEntryFilterHolder.CreateInstance(memPtr, _ignite));
}
- private int CacheEntryFilterApply(void* target, long objPtr, long memPtr)
+ private long CacheEntryFilterApply(long memPtr)
{
- return SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var t = _ignite.HandleRegistry.Get<CacheEntryFilterHolder>(objPtr);
+ var t = _ignite.HandleRegistry.Get<CacheEntryFilterHolder>(stream.ReadLong());
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return t.Invoke(stream);
- }
- });
+ return t.Invoke(stream);
+ }
}
- private void CacheEntryFilterDestroy(void* target, long objPtr)
+ private long CacheEntryFilterDestroy(long objPtr)
{
- SafeCall(() => _ignite.HandleRegistry.Release(objPtr));
+ _ignite.HandleRegistry.Release(objPtr);
+
+ return 0;
}
- private void CacheInvoke(void* target, long inMemPtr, long outMemPtr)
+ private long CacheInvoke(long memPtr)
{
- SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (PlatformMemoryStream inStream = IgniteManager.Memory.Get(inMemPtr).GetStream())
- {
- var result = ReadAndRunCacheEntryProcessor(inStream, _ignite);
+ var result = ReadAndRunCacheEntryProcessor(stream, _ignite);
- using (PlatformMemoryStream outStream = IgniteManager.Memory.Get(outMemPtr).GetStream())
- {
- result.Write(outStream, _ignite.Marshaller);
+ stream.Reset();
- outStream.SynchronizeOutput();
- }
- }
- });
+ result.Write(stream, _ignite.Marshaller);
+
+ stream.SynchronizeOutput();
+ }
+
+ return 0;
}
/// <summary>
@@ -432,122 +444,110 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
#region IMPLEMENTATION: COMPUTE
- private void ComputeTaskMap(void* target, long taskPtr, long inMemPtr, long outMemPtr)
+ private long ComputeTaskMap(long memPtr)
{
- SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (PlatformMemoryStream inStream = IgniteManager.Memory.Get(inMemPtr).GetStream())
- {
- using (PlatformMemoryStream outStream = IgniteManager.Memory.Get(outMemPtr).GetStream())
- {
- Task(taskPtr).Map(inStream, outStream);
- }
- }
- });
+ Task(stream.ReadLong()).Map(stream);
+
+ return 0;
+ }
+ }
+
+ private long ComputeTaskLocalJobResult(long taskPtr, long jobPtr, long unused, void* arg)
+ {
+ return Task(taskPtr).JobResultLocal(Job(jobPtr));
}
- private int ComputeTaskJobResult(void* target, long taskPtr, long jobPtr, long memPtr)
+ private long ComputeTaskJobResult(long memPtr)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var task = Task(taskPtr);
+ var task = Task(stream.ReadLong());
- if (memPtr == 0)
- {
- return task.JobResultLocal(Job(jobPtr));
- }
+ var job = Job(stream.ReadLong());
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return task.JobResultRemote(Job(jobPtr), stream);
- }
- });
+ return task.JobResultRemote(job, stream);
+ }
}
- private void ComputeTaskReduce(void* target, long taskPtr)
+ private long ComputeTaskReduce(long taskPtr)
{
- SafeCall(() =>
- {
- var task = _handleRegistry.Get<IComputeTaskHolder>(taskPtr, true);
+ _handleRegistry.Get<IComputeTaskHolder>(taskPtr, true).Reduce();
- task.Reduce();
- });
+ return 0;
}
- private void ComputeTaskComplete(void* target, long taskPtr, long memPtr)
+ private long ComputeTaskComplete(long taskPtr, long memPtr, long unused, void* arg)
{
- SafeCall(() =>
- {
- var task = _handleRegistry.Get<IComputeTaskHolder>(taskPtr, true);
+ var task = _handleRegistry.Get<IComputeTaskHolder>(taskPtr, true);
- if (memPtr == 0)
- task.Complete(taskPtr);
- else
+ if (memPtr == 0)
+ task.Complete(taskPtr);
+ else
+ {
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- task.CompleteWithError(taskPtr, stream);
- }
+ task.CompleteWithError(taskPtr, stream);
}
- });
+ }
+
+ return 0;
}
- private int ComputeJobSerialize(void* target, long jobPtr, long memPtr)
+ private long ComputeJobSerialize(long jobPtr, long memPtr, long unused, void* arg)
{
- return SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return Job(jobPtr).Serialize(stream) ? 1 : 0;
- }
- });
+ return Job(jobPtr).Serialize(stream) ? 1 : 0;
+ }
}
- private long ComputeJobCreate(void* target, long memPtr)
+ private long ComputeJobCreate(long memPtr)
{
- return SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- ComputeJobHolder job = ComputeJobHolder.CreateJob(_ignite, stream);
+ ComputeJobHolder job = ComputeJobHolder.CreateJob(_ignite, stream);
- return _handleRegistry.Allocate(job);
- }
- });
+ return _handleRegistry.Allocate(job);
+ }
}
- private void ComputeJobExecute(void* target, long jobPtr, int cancel, long memPtr)
+ private long ComputeJobExecuteLocal(long jobPtr, long cancel, long unused, void* arg)
{
- SafeCall(() =>
- {
- var job = Job(jobPtr);
+ Job(jobPtr).ExecuteLocal(cancel == 1);
- if (memPtr == 0)
- job.ExecuteLocal(cancel == 1);
- else
- {
- using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- job.ExecuteRemote(stream, cancel == 1);
- }
- }
- });
+ return 0;
}
- private void ComputeJobCancel(void* target, long jobPtr)
+ private long ComputeJobExecute(long memPtr)
{
- SafeCall(() =>
+ using (PlatformMemoryStream stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- Job(jobPtr).Cancel();
- });
+ var job = Job(stream.ReadLong());
+
+ var cancel = stream.ReadBool();
+
+ stream.Reset();
+
+ job.ExecuteRemote(stream, cancel);
+ }
+
+ return 0;
}
- private void ComputeJobDestroy(void* target, long jobPtr)
+ private long ComputeJobCancel(long jobPtr)
{
- SafeCall(() =>
- {
- _handleRegistry.Release(jobPtr);
- });
+ Job(jobPtr).Cancel();
+
+ return 0;
+ }
+
+ private long ComputeJobDestroy(long jobPtr)
+ {
+ _handleRegistry.Release(jobPtr);
+
+ return 0;
}
/// <summary>
@@ -574,219 +574,178 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
#region IMPLEMENTATION: CONTINUOUS QUERY
- private void ContinuousQueryListenerApply(void* target, long lsnrPtr, long memPtr)
+ private long ContinuousQueryListenerApply(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var hnd = _handleRegistry.Get<IContinuousQueryHandleImpl>(lsnrPtr);
+ var hnd = _handleRegistry.Get<IContinuousQueryHandleImpl>(stream.ReadLong());
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- hnd.Apply(stream);
- }
- });
+ hnd.Apply(stream);
+
+ return 0;
+ }
}
[SuppressMessage("ReSharper", "PossibleNullReferenceException")]
- private long ContinuousQueryFilterCreate(void* target, long memPtr)
+ private long ContinuousQueryFilterCreate(long memPtr)
{
- return SafeCall(() =>
+ // 1. Unmarshal filter holder.
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- // 1. Unmarshal filter holder.
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- var filterHolder = reader.ReadObject<ContinuousQueryFilterHolder>();
+ var filterHolder = reader.ReadObject<ContinuousQueryFilterHolder>();
- // 2. Create real filter from it's holder.
- var filter = (IContinuousQueryFilter)DelegateTypeDescriptor.GetContinuousQueryFilterCtor(
- filterHolder.Filter.GetType())(filterHolder.Filter, filterHolder.KeepBinary);
+ // 2. Create real filter from it's holder.
+ var filter = (IContinuousQueryFilter) DelegateTypeDescriptor.GetContinuousQueryFilterCtor(
+ filterHolder.Filter.GetType())(filterHolder.Filter, filterHolder.KeepBinary);
- // 3. Inject grid.
- filter.Inject(_ignite);
+ // 3. Inject grid.
+ filter.Inject(_ignite);
- // 4. Allocate GC handle.
- return filter.Allocate();
- }
- });
+ // 4. Allocate GC handle.
+ return filter.Allocate();
+ }
}
- private int ContinuousQueryFilterApply(void* target, long filterPtr, long memPtr)
+ private long ContinuousQueryFilterApply(long memPtr)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var holder = _handleRegistry.Get<IContinuousQueryFilter>(filterPtr);
+ var holder = _handleRegistry.Get<IContinuousQueryFilter>(stream.ReadLong());
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return holder.Evaluate(stream) ? 1 : 0;
- }
- });
+ return holder.Evaluate(stream) ? 1 : 0;
+ }
}
- private void ContinuousQueryFilterRelease(void* target, long filterPtr)
+ private long ContinuousQueryFilterRelease(long filterPtr)
{
- SafeCall(() =>
- {
- var holder = _handleRegistry.Get<IContinuousQueryFilter>(filterPtr);
+ var holder = _handleRegistry.Get<IContinuousQueryFilter>(filterPtr);
- holder.Release();
- });
+ holder.Release();
+
+ return 0;
}
#endregion
#region IMPLEMENTATION: DATA STREAMER
- private void DataStreamerTopologyUpdate(void* target, long ldrPtr, long topVer, int topSize)
+ private long DataStreamerTopologyUpdate(long ldrPtr, long topVer, long topSize, void* unused)
{
- SafeCall(() =>
- {
- var ldrRef = _handleRegistry.Get<WeakReference>(ldrPtr);
+ var ldrRef = _handleRegistry.Get<WeakReference>(ldrPtr);
- if (ldrRef == null)
- return;
+ if (ldrRef == null)
+ return 0;
- var ldr = ldrRef.Target as IDataStreamer;
+ var ldr = ldrRef.Target as IDataStreamer;
- if (ldr != null)
- ldr.TopologyChange(topVer, topSize);
- else
- _handleRegistry.Release(ldrPtr, true);
- });
+ if (ldr != null)
+ ldr.TopologyChange(topVer, (int) topSize);
+ else
+ _handleRegistry.Release(ldrPtr, true);
+
+ return 0;
}
[SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope")]
- private void DataStreamerStreamReceiverInvoke(void* target, long rcvPtr, void* cache, long memPtr,
- byte keepBinary)
+ private long DataStreamerStreamReceiverInvoke(long memPtr, long unused, long unused1, void* cache)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream, BinaryMode.ForceBinary);
+ var rcvPtr = stream.ReadLong();
- var binaryReceiver = reader.ReadObject<BinaryObject>();
+ var keepBinary = stream.ReadBool();
- var receiver = _handleRegistry.Get<StreamReceiverHolder>(rcvPtr) ??
- binaryReceiver.Deserialize<StreamReceiverHolder>();
+ var reader = _ignite.Marshaller.StartUnmarshal(stream, BinaryMode.ForceBinary);
- if (receiver != null)
- receiver.Receive(_ignite, new UnmanagedNonReleaseableTarget(_ctx, cache), stream,
- keepBinary != 0);
- }
- });
+ var binaryReceiver = reader.ReadObject<BinaryObject>();
+
+ var receiver = _handleRegistry.Get<StreamReceiverHolder>(rcvPtr) ??
+ binaryReceiver.Deserialize<StreamReceiverHolder>();
+
+ if (receiver != null)
+ receiver.Receive(_ignite, new UnmanagedNonReleaseableTarget(_ctx, cache), stream, keepBinary);
+
+ return 0;
+ }
}
#endregion
#region IMPLEMENTATION: FUTURES
- private void FutureByteResult(void* target, long futPtr, int res)
+ private long FutureByteResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<byte>(futPtr, fut => { fut.OnResult((byte)res); });
- });
+ return ProcessFuture<byte>(futPtr, fut => { fut.OnResult((byte) res); });
}
- private void FutureBoolResult(void* target, long futPtr, int res)
+ private long FutureBoolResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<bool>(futPtr, fut => { fut.OnResult(res == 1); });
- });
+ return ProcessFuture<bool>(futPtr, fut => { fut.OnResult(res == 1); });
}
- private void FutureShortResult(void* target, long futPtr, int res)
+ private long FutureShortResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<short>(futPtr, fut => { fut.OnResult((short)res); });
- });
+ return ProcessFuture<short>(futPtr, fut => { fut.OnResult((short)res); });
}
- private void FutureCharResult(void* target, long futPtr, int res)
+ private long FutureCharResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<char>(futPtr, fut => { fut.OnResult((char)res); });
- });
+ return ProcessFuture<char>(futPtr, fut => { fut.OnResult((char)res); });
}
- private void FutureIntResult(void* target, long futPtr, int res)
+ private long FutureIntResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<int>(futPtr, fut => { fut.OnResult(res); });
- });
+ return ProcessFuture<int>(futPtr, fut => { fut.OnResult((int) res); });
}
- private void FutureFloatResult(void* target, long futPtr, float res)
+ private long FutureFloatResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<float>(futPtr, fut => { fut.OnResult(res); });
- });
+ return ProcessFuture<float>(futPtr, fut => { fut.OnResult(BinaryUtils.IntToFloatBits((int) res)); });
}
- private void FutureLongResult(void* target, long futPtr, long res)
+ private long FutureLongResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<long>(futPtr, fut => { fut.OnResult(res); });
- });
+ return ProcessFuture<long>(futPtr, fut => { fut.OnResult(res); });
}
- private void FutureDoubleResult(void* target, long futPtr, double res)
+ private long FutureDoubleResult(long futPtr, long res, long unused, void* arg)
{
- SafeCall(() =>
- {
- ProcessFuture<double>(futPtr, fut => { fut.OnResult(res); });
- });
+ return ProcessFuture<double>(futPtr, fut => { fut.OnResult(BinaryUtils.LongToDoubleBits(res)); });
}
- private void FutureObjectResult(void* target, long futPtr, long memPtr)
+ private long FutureObjectResult(long futPtr, long memPtr, long unused, void* arg)
{
- SafeCall(() =>
+ return ProcessFuture(futPtr, fut =>
{
- ProcessFuture(futPtr, fut =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- fut.OnResult(stream);
- }
- });
+ fut.OnResult(stream);
+ }
});
}
- private void FutureNullResult(void* target, long futPtr)
+ private long FutureNullResult(long futPtr)
{
- SafeCall(() =>
- {
- ProcessFuture(futPtr, fut => { fut.OnNullResult(); });
- });
+ return ProcessFuture(futPtr, fut => { fut.OnNullResult(); });
}
- private void FutureError(void* target, long futPtr, long memPtr)
+ private long FutureError(long futPtr, long memPtr, long unused, void* arg)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- string errCls = reader.ReadString();
- string errMsg = reader.ReadString();
- string stackTrace = reader.ReadString();
- Exception inner = reader.ReadBoolean() ? reader.ReadObject<Exception>() : null;
+ string errCls = reader.ReadString();
+ string errMsg = reader.ReadString();
+ string stackTrace = reader.ReadString();
+ Exception inner = reader.ReadBoolean() ? reader.ReadObject<Exception>() : null;
- Exception err = ExceptionUtils.GetException(_ignite, errCls, errMsg, stackTrace, reader, inner);
+ Exception err = ExceptionUtils.GetException(_ignite, errCls, errMsg, stackTrace, reader, inner);
- ProcessFuture(futPtr, fut => { fut.OnError(err); });
- }
- });
+ return ProcessFuture(futPtr, fut => { fut.OnError(err); });
+ }
}
/// <summary>
@@ -794,11 +753,13 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
/// </summary>
/// <param name="futPtr">Future pointer.</param>
/// <param name="action">Action.</param>
- private void ProcessFuture(long futPtr, Action<IFutureInternal> action)
+ private long ProcessFuture(long futPtr, Action<IFutureInternal> action)
{
try
{
action(_handleRegistry.Get<IFutureInternal>(futPtr, true));
+
+ return 0;
}
finally
{
@@ -811,11 +772,13 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
/// </summary>
/// <param name="futPtr">Future pointer.</param>
/// <param name="action">Action.</param>
- private void ProcessFuture<T>(long futPtr, Action<Future<T>> action)
+ private long ProcessFuture<T>(long futPtr, Action<Future<T>> action)
{
try
{
action(_handleRegistry.Get<Future<T>>(futPtr, true));
+
+ return 0;
}
finally
{
@@ -827,257 +790,230 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
#region IMPLEMENTATION: LIFECYCLE
- private void LifecycleOnEvent(void* target, long ptr, int evt)
+ private long LifecycleOnEvent(long ptr, long evt, long unused, void* arg)
{
- SafeCall(() =>
- {
- var bean = _handleRegistry.Get<LifecycleBeanHolder>(ptr);
+ var bean = _handleRegistry.Get<LifecycleBeanHolder>(ptr);
- bean.OnLifecycleEvent((LifecycleEventType)evt);
- }, true);
+ bean.OnLifecycleEvent((LifecycleEventType) evt);
+
+ return 0;
}
#endregion
#region IMPLEMENTATION: MESSAGING
- private long MessagingFilterCreate(void* target, long memPtr)
+ private long MessagingFilterCreate(long memPtr)
{
- return SafeCall(() =>
- {
- MessageListenerHolder holder = MessageListenerHolder.CreateRemote(_ignite, memPtr);
+ MessageListenerHolder holder = MessageListenerHolder.CreateRemote(_ignite, memPtr);
- return _ignite.HandleRegistry.AllocateSafe(holder);
- });
+ return _ignite.HandleRegistry.AllocateSafe(holder);
}
- private int MessagingFilterApply(void* target, long ptr, long memPtr)
+ private long MessagingFilterApply(long ptr, long memPtr, long unused, void* arg)
{
- return SafeCall(() =>
- {
- var holder = _ignite.HandleRegistry.Get<MessageListenerHolder>(ptr, false);
+ var holder = _ignite.HandleRegistry.Get<MessageListenerHolder>(ptr, false);
- if (holder == null)
- return 0;
+ if (holder == null)
+ return 0;
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return holder.Invoke(stream);
- }
- });
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
+ {
+ return holder.Invoke(stream);
+ }
}
- private void MessagingFilterDestroy(void* target, long ptr)
+ private long MessagingFilterDestroy(long ptr)
{
- SafeCall(() =>
- {
- _ignite.HandleRegistry.Release(ptr);
- });
+ _ignite.HandleRegistry.Release(ptr);
+
+ return 0;
}
#endregion
#region IMPLEMENTATION: EXTENSIONS
- private long ExtensionCallbackInLongOutLong(void* target, int op, long arg1)
+ private long ExtensionCallbackInLongLongOutLong(long op, long arg1, long arg2, void* arg)
{
- throw new InvalidOperationException("Unsupported operation type: " + op);
- }
-
- private long ExtensionCallbackInLongLongOutLong(void* target, int op, long arg1, long arg2)
- {
- return SafeCall(() =>
+ switch (op)
{
- switch (op)
- {
- case OpPrepareDotNet:
- using (var inStream = IgniteManager.Memory.Get(arg1).GetStream())
- using (var outStream = IgniteManager.Memory.Get(arg2).GetStream())
- {
- Ignition.OnPrepare(inStream, outStream, _handleRegistry, _log);
+ case OpPrepareDotNet:
+ using (var inStream = IgniteManager.Memory.Get(arg1).GetStream())
+ using (var outStream = IgniteManager.Memory.Get(arg2).GetStream())
+ {
+ Ignition.OnPrepare(inStream, outStream, _handleRegistry, _log);
- return 0;
- }
+ return 0;
+ }
- default:
- throw new InvalidOperationException("Unsupported operation type: " + op);
- }
- }, op == OpPrepareDotNet);
+ default:
+ throw new InvalidOperationException("Unsupported operation type: " + op);
+ }
}
#endregion
#region IMPLEMENTATION: EVENTS
- private long EventFilterCreate(void* target, long memPtr)
+ private long EventFilterCreate(long memPtr)
{
- return SafeCall(() => _handleRegistry.Allocate(RemoteListenEventFilter.CreateInstance(memPtr, _ignite)));
+ return _handleRegistry.Allocate(RemoteListenEventFilter.CreateInstance(memPtr, _ignite));
}
- private int EventFilterApply(void* target, long ptr, long memPtr)
+ private long EventFilterApply(long ptr, long memPtr, long unused, void* arg)
{
- return SafeCall(() =>
- {
- var holder = _ignite.HandleRegistry.Get<IInteropCallback>(ptr, false);
+ var holder = _ignite.HandleRegistry.Get<IInteropCallback>(ptr, false);
- if (holder == null)
- return 0;
+ if (holder == null)
+ return 0;
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- return holder.Invoke(stream);
- }
- });
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
+ {
+ return holder.Invoke(stream);
+ }
}
- private void EventFilterDestroy(void* target, long ptr)
+ private long EventFilterDestroy(long ptr)
{
- SafeCall(() =>
- {
- _ignite.HandleRegistry.Release(ptr);
- });
+ _ignite.HandleRegistry.Release(ptr);
+
+ return 0;
}
#endregion
#region IMPLEMENTATION: SERVICES
- private long ServiceInit(void* target, long memPtr)
+ private long ServiceInit(long memPtr)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- bool srvKeepBinary = reader.ReadBoolean();
- var svc = reader.ReadObject<IService>();
+ bool srvKeepBinary = reader.ReadBoolean();
+ var svc = reader.ReadObject<IService>();
- ResourceProcessor.Inject(svc, _ignite);
+ ResourceProcessor.Inject(svc, _ignite);
- svc.Init(new ServiceContext(_ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
+ svc.Init(new ServiceContext(_ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
- return _handleRegistry.Allocate(svc);
- }
- });
+ return _handleRegistry.Allocate(svc);
+ }
}
- private void ServiceExecute(void* target, long svcPtr, long memPtr)
+ private long ServiceExecute(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var svc = _handleRegistry.Get<IService>(svcPtr);
+ var svc = _handleRegistry.Get<IService>(stream.ReadLong());
// Ignite does not guarantee that Cancel is called after Execute exits
// So missing handle is a valid situation
if (svc == null)
- return;
+ return 0;
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- bool srvKeepBinary = reader.ReadBoolean();
+ bool srvKeepBinary = reader.ReadBoolean();
- svc.Execute(new ServiceContext(
- _ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
- }
- });
+ svc.Execute(new ServiceContext(_ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
+
+ return 0;
+ }
}
- private void ServiceCancel(void* target, long svcPtr, long memPtr)
+ private long ServiceCancel(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var svc = _handleRegistry.Get<IService>(svcPtr, true);
+ long svcPtr = stream.ReadLong();
try
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var svc = _handleRegistry.Get<IService>(svcPtr, true);
- bool srvKeepBinary = reader.ReadBoolean();
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- svc.Cancel(new ServiceContext(_ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
- }
+ bool srvKeepBinary = reader.ReadBoolean();
+
+ svc.Cancel(new ServiceContext(_ignite.Marshaller.StartUnmarshal(stream, srvKeepBinary)));
+
+ return 0;
}
finally
{
_ignite.HandleRegistry.Release(svcPtr);
}
- });
+ }
}
- private void ServiceInvokeMethod(void* target, long svcPtr, long inMemPtr, long outMemPtr)
+ private long ServiceInvokeMethod(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var inStream = IgniteManager.Memory.Get(inMemPtr).GetStream())
- using (var outStream = IgniteManager.Memory.Get(outMemPtr).GetStream())
- {
- var svc = _handleRegistry.Get<IService>(svcPtr, true);
+ var svc = _handleRegistry.Get<IService>(stream.ReadLong(), true);
- string mthdName;
- object[] mthdArgs;
+ string mthdName;
+ object[] mthdArgs;
- ServiceProxySerializer.ReadProxyMethod(inStream, _ignite.Marshaller, out mthdName, out mthdArgs);
+ ServiceProxySerializer.ReadProxyMethod(stream, _ignite.Marshaller, out mthdName, out mthdArgs);
- var result = ServiceProxyInvoker.InvokeServiceMethod(svc, mthdName, mthdArgs);
+ var result = ServiceProxyInvoker.InvokeServiceMethod(svc, mthdName, mthdArgs);
- ServiceProxySerializer.WriteInvocationResult(outStream, _ignite.Marshaller, result.Key, result.Value);
+ stream.Reset();
- outStream.SynchronizeOutput();
- }
- });
+ ServiceProxySerializer.WriteInvocationResult(stream, _ignite.Marshaller, result.Key, result.Value);
+
+ stream.SynchronizeOutput();
+
+ return 0;
+ }
}
- private int ClusterNodeFilterApply(void* target, long memPtr)
+ private long ClusterNodeFilterApply(long memPtr)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- var filter = reader.ReadObject<IClusterNodeFilter>();
+ var filter = reader.ReadObject<IClusterNodeFilter>();
- return filter.Invoke(_ignite.GetNode(reader.ReadGuid())) ? 1 : 0;
- }
- });
+ return filter.Invoke(_ignite.GetNode(reader.ReadGuid())) ? 1 : 0;
+ }
}
#endregion
#region IMPLEMENTATION: MISCELLANEOUS
- private void NodeInfo(void* target, long memPtr)
+ private long NodeInfo(long memPtr)
{
- SafeCall(() => _ignite.UpdateNodeInfo(memPtr));
+ _ignite.UpdateNodeInfo(memPtr);
+
+ return 0;
}
- private void MemoryReallocate(void* target, long memPtr, int cap)
+ private long MemoryReallocate(long memPtr, long cap, long unused, void* arg)
{
- SafeCall(() =>
- {
- IgniteManager.Memory.Get(memPtr).Reallocate(cap);
- }, true);
+ IgniteManager.Memory.Get(memPtr).Reallocate((int)cap);
+
+ return 0;
}
- private void OnStart(void* target, void* proc, long memPtr)
+ private long OnStart(long memPtr, long unused, long unused1, void* proc)
{
- SafeCall(() =>
+ var proc0 = UU.Acquire(_ctx, proc);
+
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- var proc0 = UnmanagedUtils.Acquire(_ctx, proc);
+ Ignition.OnStart(proc0, stream);
+ }
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- Ignition.OnStart(proc0, stream);
- }
- }, true);
+ return 0;
}
- private void OnStop(void* target)
+ private long OnStop(long unused)
{
Marshal.FreeHGlobal(_cbsPtr);
@@ -1092,6 +1028,8 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
if (ignite != null)
ignite.AfterNodeStop();
+
+ return 0;
}
private void Error(void* target, int errType, sbyte* errClsChars, int errClsCharsLen, sbyte* errMsgChars,
@@ -1122,20 +1060,18 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
}
}
- private void OnClientDisconnected(void* target)
+ private long OnClientDisconnected(long unused)
{
- SafeCall(() =>
- {
- _ignite.OnClientDisconnected();
- });
+ _ignite.OnClientDisconnected();
+
+ return 0;
}
- private void OnClientReconnected(void* target, bool clusterRestarted)
+ private long OnClientReconnected(long clusterRestarted)
{
- SafeCall(() =>
- {
- _ignite.OnClientReconnected(clusterRestarted);
- });
+ _ignite.OnClientReconnected(clusterRestarted != 0);
+
+ return 0;
}
private void LoggerLog(void* target, int level, sbyte* messageChars, int messageCharsLen, sbyte* categoryChars,
@@ -1194,82 +1130,79 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
#region AffinityFunction
- private long AffinityFunctionInit(void* target, long memPtr, void* baseFunc)
+ private long AffinityFunctionInit(long memPtr, long unused, long unused1, void* baseFunc)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var reader = _ignite.Marshaller.StartUnmarshal(stream);
+ var reader = _ignite.Marshaller.StartUnmarshal(stream);
- var func = reader.ReadObjectEx<IAffinityFunction>();
+ var func = reader.ReadObjectEx<IAffinityFunction>();
- ResourceProcessor.Inject(func, _ignite);
+ ResourceProcessor.Inject(func, _ignite);
- var affBase = func as AffinityFunctionBase;
+ var affBase = func as AffinityFunctionBase;
- if (affBase != null)
- affBase.SetBaseFunction(new PlatformAffinityFunction(
- _ignite.InteropProcessor.ChangeTarget(baseFunc), _ignite.Marshaller));
+ if (affBase != null)
+ {
+ var baseFunc0 = UU.Acquire(_ctx, baseFunc);
- return _handleRegistry.Allocate(func);
+ affBase.SetBaseFunction(new PlatformAffinityFunction(baseFunc0, _ignite.Marshaller));
}
- });
+
+ return _handleRegistry.Allocate(func);
+ }
}
- private int AffinityFunctionPartition(void* target, long ptr, long memPtr)
+ private long AffinityFunctionPartition(long memPtr)
{
- return SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var key = _ignite.Marshaller.Unmarshal<object>(stream);
+ var ptr = stream.ReadLong();
- return _handleRegistry.Get<IAffinityFunction>(ptr, true).GetPartition(key);
- }
- });
+ var key = _ignite.Marshaller.Unmarshal<object>(stream);
+
+ return _handleRegistry.Get<IAffinityFunction>(ptr, true).GetPartition(key);
+ }
}
- private void AffinityFunctionAssignPartitions(void* target, long ptr, long inMemPtr, long outMemPtr)
+ private long AffinityFunctionAssignPartitions(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var inStream = IgniteManager.Memory.Get(inMemPtr).GetStream())
- {
- var ctx = new AffinityFunctionContext(_ignite.Marshaller.StartUnmarshal(inStream));
- var func = _handleRegistry.Get<IAffinityFunction>(ptr, true);
- var parts = func.AssignPartitions(ctx);
+ var ptr = stream.ReadLong();
+ var ctx = new AffinityFunctionContext(_ignite.Marshaller.StartUnmarshal(stream));
+ var func = _handleRegistry.Get<IAffinityFunction>(ptr, true);
+ var parts = func.AssignPartitions(ctx);
- if (parts == null)
- throw new IgniteException(func.GetType() + ".AssignPartitions() returned invalid result: null");
+ if (parts == null)
+ throw new IgniteException(func.GetType() + ".AssignPartitions() returned invalid result: null");
- using (var outStream = IgniteManager.Memory.Get(outMemPtr).GetStream())
- {
- AffinityFunctionSerializer.WritePartitions(parts, outStream, _ignite.Marshaller);
- }
- }
- });
+ stream.Reset();
+
+ AffinityFunctionSerializer.WritePartitions(parts, stream, _ignite.Marshaller);
+
+ return 0;
+ }
}
- private void AffinityFunctionRemoveNode(void* target, long ptr, long memPtr)
+ private long AffinityFunctionRemoveNode(long memPtr)
{
- SafeCall(() =>
+ using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
{
- using (var stream = IgniteManager.Memory.Get(memPtr).GetStream())
- {
- var nodeId = _ignite.Marshaller.Unmarshal<Guid>(stream);
+ var ptr = stream.ReadLong();
+ var nodeId = _ignite.Marshaller.Unmarshal<Guid>(stream);
- _handleRegistry.Get<IAffinityFunction>(ptr, true).RemoveNode(nodeId);
- }
- });
+ _handleRegistry.Get<IAffinityFunction>(ptr, true).RemoveNode(nodeId);
+
+ return 0;
+ }
}
- private void AffinityFunctionDestroy(void* target, long ptr)
+ private long AffinityFunctionDestroy(long ptr)
{
- SafeCall(() =>
- {
- _handleRegistry.Release(ptr);
- });
+ _handleRegistry.Release(ptr);
+
+ return 0;
}
#endregion
@@ -1315,7 +1248,7 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
}
#endregion
-
+
/// <summary>
/// Callbacks pointer.
/// </summary>
@@ -1398,5 +1331,47 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
{
get { return ConsoleWritePtr; }
}
+
+ /// <summary>
+ /// InLongOutLong handler struct.
+ /// </summary>
+ private struct InLongOutLongHandler
+ {
+ /// <summary> The handler func. </summary>
+ public readonly InLongOutLongFunc Handler;
+
+ /// <summary> Allow uninitialized flag. </summary>
+ public readonly bool AllowUninitialized;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="InLongOutLongHandler"/> struct.
+ /// </summary>
+ public InLongOutLongHandler(InLongOutLongFunc handler, bool allowUninitialized)
+ {
+ Handler = handler;
+ AllowUninitialized = allowUninitialized;
+ }
+ }
+
+ /// <summary>
+ /// InLongLongLongObjectOutLong handler struct.
+ /// </summary>
+ private struct InLongLongLongObjectOutLongHandler
+ {
+ /// <summary> The handler func. </summary>
+ public readonly InLongLongLongObjectOutLongFunc Handler;
+
+ /// <summary> Allow uninitialized flag. </summary>
+ public readonly bool AllowUninitialized;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="InLongLongLongObjectOutLongHandler"/> struct.
+ /// </summary>
+ public InLongLongLongObjectOutLongHandler(InLongLongLongObjectOutLongFunc handler, bool allowUninitialized)
+ {
+ Handler = handler;
+ AllowUninitialized = allowUninitialized;
+ }
+ }
}
}
\ No newline at end of file
[2/3] ignite git commit: IGNITE-4033 Streamline platform callback
interface
Posted by pt...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/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 32c855b..4947b94 100644
--- a/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
+++ b/modules/platforms/cpp/core/src/impl/ignite_environment.cpp
@@ -29,48 +29,66 @@ using namespace ignite::binary;
namespace ignite
{
- namespace impl
+ namespace impl
{
/**
- * OnStart callback.
- *
- * @param target Target environment.
- * @param proc Processor instance.
- * @param memPtr Memory pointer.
- */
- void IGNITE_CALL OnStart(void* target, void* proc, long long memPtr)
+ * Callback codes.
+ */
+ enum CallbackOp
{
- SharedPointer<IgniteEnvironment>* ptr = static_cast<SharedPointer<IgniteEnvironment>*>(target);
-
- ptr->Get()->OnStartCallback(memPtr, reinterpret_cast<jobject>(proc));
- }
+ REALLOC = 36,
+ ON_START = 49,
+ ON_STOP = 50
+ };
/**
- * OnStop callback.
- *
+ * InLongOutLong callback.
+ *
* @param target Target environment.
+ * @param type Operation type.
+ * @param val Value.
*/
- void IGNITE_CALL OnStop(void* target)
+ long long IGNITE_CALL InLongOutLong(void* target, int type, long long val)
{
- SharedPointer<IgniteEnvironment>* ptr = static_cast<SharedPointer<IgniteEnvironment>*>(target);
+ if (type == ON_STOP)
+ {
+ SharedPointer<IgniteEnvironment>* ptr = static_cast<SharedPointer<IgniteEnvironment>*>(target);
+
+ delete ptr;
+ }
- delete ptr;
+ return 0;
}
/**
- * Memory reallocate callback.
- *
+ * InLongOutLong callback.
+ *
* @param target Target environment.
- * @param memPtr Memory pointer.
- * @param cap Required capasity.
+ * @param type Operation type.
+ * @param val1 Value1.
+ * @param val2 Value2.
+ * @param val3 Value3.
+ * @param arg Object arg.
*/
- void IGNITE_CALL MemoryReallocate(void* target, long long memPtr, int cap)
+ long long IGNITE_CALL InLongLongLongObjectOutLong(void* target, int type, long long val1, long long val2,
+ long long val3, void* arg)
{
- SharedPointer<IgniteEnvironment>* env = static_cast<SharedPointer<IgniteEnvironment>*>(target);
+ if (type == ON_START)
+ {
+ SharedPointer<IgniteEnvironment>* ptr = static_cast<SharedPointer<IgniteEnvironment>*>(target);
+
+ ptr->Get()->OnStartCallback(val1, reinterpret_cast<jobject>(arg));
+ }
+ else if (type == REALLOC)
+ {
+ SharedPointer<IgniteEnvironment>* env = static_cast<SharedPointer<IgniteEnvironment>*>(target);
- SharedPointer<InteropMemory> mem = env->Get()->GetMemory(memPtr);
+ SharedPointer<InteropMemory> mem = env->Get()->GetMemory(val1);
- mem.Get()->Reallocate(cap);
+ mem.Get()->Reallocate(static_cast<int32_t>(val2));
+ }
+
+ return 0;
}
IgniteEnvironment::IgniteEnvironment() : ctx(SharedPointer<JniContext>()), latch(new SingleLatch), name(0),
@@ -93,10 +111,8 @@ namespace ignite
hnds.target = target;
- hnds.onStart = OnStart;
- hnds.onStop = OnStop;
-
- hnds.memRealloc = MemoryReallocate;
+ hnds.inLongOutLong = InLongOutLong;
+ hnds.inLongLongLongObjectOutLong = InLongLongLongObjectOutLong;
hnds.error = 0;
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/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 97e4412..442cc10 100644
--- a/modules/platforms/cpp/jni/include/ignite/jni/java.h
+++ b/modules/platforms/cpp/jni/include/ignite/jni/java.h
@@ -110,6 +110,9 @@ namespace ignite
typedef void(JNICALL *LoggerLogHandler)(void* target, int level, const char* messageChars, int messageCharsLen, const char* categoryChars, int categoryCharsLen, const char* errorInfoChars, int errorInfoCharsLen, long long memPtr);
typedef bool(JNICALL *LoggerIsLevelEnabledHandler)(void* target, int level);
+
+ typedef long long(JNICALL *InLongOutLongHandler)(void* target, int type, long long val);
+ typedef long long(JNICALL *InLongLongLongObjectOutLongHandler)(void* target, int type, long long val1, long long val2, long long val3, void* arg);
/**
* JNI handlers holder.
@@ -117,86 +120,13 @@ namespace ignite
struct JniHandlers {
void* target;
- CacheStoreCreateHandler cacheStoreCreate;
- CacheStoreInvokeHandler cacheStoreInvoke;
- CacheStoreDestroyHandler cacheStoreDestroy;
- CacheStoreSessionCreateHandler cacheStoreSessionCreate;
-
- CacheEntryFilterCreateHandler cacheEntryFilterCreate;
- CacheEntryFilterApplyHandler cacheEntryFilterApply;
- CacheEntryFilterDestroyHandler cacheEntryFilterDestroy;
-
- CacheInvokeHandler cacheInvoke;
-
- ComputeTaskMapHandler computeTaskMap;
- ComputeTaskJobResultHandler computeTaskJobRes;
- ComputeTaskReduceHandler computeTaskReduce;
- ComputeTaskCompleteHandler computeTaskComplete;
- ComputeJobSerializeHandler computeJobSerialize;
- ComputeJobCreateHandler computeJobCreate;
- ComputeJobExecuteHandler computeJobExec;
- ComputeJobCancelHandler computeJobCancel;
- ComputeJobDestroyHandler computeJobDestroy;
-
- ContinuousQueryListenerApplyHandler contQryLsnrApply;
- ContinuousQueryFilterCreateHandler contQryFilterCreate;
- ContinuousQueryFilterApplyHandler contQryFilterApply;
- ContinuousQueryFilterReleaseHandler contQryFilterRelease;
-
- DataStreamerTopologyUpdateHandler dataStreamerTopologyUpdate;
- DataStreamerStreamReceiverInvokeHandler streamReceiverInvoke;
-
- FutureByteResultHandler futByteRes;
- FutureBoolResultHandler futBoolRes;
- FutureShortResultHandler futShortRes;
- FutureCharResultHandler futCharRes;
- FutureIntResultHandler futIntRes;
- FutureFloatResultHandler futFloatRes;
- FutureLongResultHandler futLongRes;
- FutureDoubleResultHandler futDoubleRes;
- FutureObjectResultHandler futObjRes;
- FutureNullResultHandler futNullRes;
- FutureErrorHandler futErr;
-
- LifecycleEventHandler lifecycleEvt;
-
- MemoryReallocateHandler memRealloc;
-
- MessagingFilterCreateHandler messagingFilterCreate;
- MessagingFilterApplyHandler messagingFilterApply;
- MessagingFilterDestroyHandler messagingFilterDestroy;
-
- EventFilterCreateHandler eventFilterCreate;
- EventFilterApplyHandler eventFilterApply;
- EventFilterDestroyHandler eventFilterDestroy;
-
- ServiceInitHandler serviceInit;
- ServiceExecuteHandler serviceExecute;
- ServiceCancelHandler serviceCancel;
- ServiceInvokeMethodHandler serviceInvokeMethod;
-
- ClusterNodeFilterApplyHandler clusterNodeFilterApply;
-
- NodeInfoHandler nodeInfo;
-
- OnStartHandler onStart;
- OnStopHandler onStop;
ErrorHandler error;
- ExtensionCallbackInLongOutLongHandler extensionCallbackInLongOutLong;
- ExtensionCallbackInLongLongOutLongHandler extensionCallbackInLongLongOutLong;
-
- OnClientDisconnectedHandler onClientDisconnected;
- OnClientReconnectedHandler onClientReconnected;
-
- AffinityFunctionInitHandler affinityFunctionInit;
- AffinityFunctionPartitionHandler affinityFunctionPartition;
- AffinityFunctionAssignPartitionsHandler affinityFunctionAssignPartitions;
- AffinityFunctionRemoveNodeHandler affinityFunctionRemoveNode;
- AffinityFunctionDestroyHandler affinityFunctionDestroy;
-
LoggerLogHandler loggerLog;
LoggerIsLevelEnabledHandler loggerIsLevelEnabled;
+
+ InLongOutLongHandler inLongOutLong;
+ InLongLongLongObjectOutLongHandler inLongLongLongObjectOutLong;
};
/**
@@ -560,6 +490,9 @@ namespace ignite
JNIEXPORT void JNICALL JniLoggerLog(JNIEnv *env, jclass cls, jlong envPtr, jint level, jstring message, jstring category, jstring errorInfo, jlong memPtr);
JNIEXPORT jboolean JNICALL JniLoggerIsLevelEnabled(JNIEnv *env, jclass cls, jlong envPtr, jint level);
+
+ JNIEXPORT jlong JNICALL JniInLongOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint type, jlong val);
+ JNIEXPORT jlong JNICALL JniInLongLongLongObjectOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint type, jlong val1, jlong val2, jlong val3, jobject arg);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/cpp/jni/project/vs/module.def
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/jni/project/vs/module.def b/modules/platforms/cpp/jni/project/vs/module.def
index fb56dca..9effaf9 100644
--- a/modules/platforms/cpp/jni/project/vs/module.def
+++ b/modules/platforms/cpp/jni/project/vs/module.def
@@ -26,8 +26,8 @@ IgniteTargetInLongOutLong @24
IgniteProcessorCompute @64
IgniteProcessorMessage @65
IgniteProcessorEvents @66
-IgniteAcquire @80
-IgniteRelease @81
+IgniteAcquire @80
+IgniteRelease @81
IgniteThrowToJava @82
IgniteHandlersSize @83
IgniteCreateContext @84
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/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 9626fbb..698263d 100644
--- a/modules/platforms/cpp/jni/src/java.cpp
+++ b/modules/platforms/cpp/jni/src/java.cpp
@@ -15,6 +15,7 @@
* limitations under the License.
*/
+// ReSharper disable once CppUnusedIncludeDirective
#include <cstring> // needed only on linux
#include <string>
#include <exception>
@@ -315,6 +316,9 @@ namespace ignite
JniMethod M_PLATFORM_CALLBACK_UTILS_CONSOLE_WRITE = JniMethod("consoleWrite", "(Ljava/lang/String;Z)V", true);
+ JniMethod M_PLATFORM_CALLBACK_UTILS_IN_LONG_OUT_LONG = JniMethod("inLongOutLong", "(JIJ)J", true);
+ JniMethod M_PLATFORM_CALLBACK_UTILS_IN_LONG_LONG_LONG_OBJECT_OUT_LONG = JniMethod("inLongLongLongObjectOutLong", "(JIJJJLjava/lang/Object;)J", true);
+
const char* C_PLATFORM_UTILS = "org/apache/ignite/internal/processors/platform/utils/PlatformUtils";
JniMethod M_PLATFORM_UTILS_REALLOC = JniMethod("reallocate", "(JI)V", true);
JniMethod M_PLATFORM_UTILS_ERR_DATA = JniMethod("errorData", "(Ljava/lang/Throwable;)[B", true);
@@ -611,93 +615,18 @@ namespace ignite
void RegisterNatives(JNIEnv* env) {
{
- JNINativeMethod methods[62];
+ JNINativeMethod methods[5];
int idx = 0;
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_STORE_CREATE, reinterpret_cast<void*>(JniCacheStoreCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_STORE_INVOKE, reinterpret_cast<void*>(JniCacheStoreInvoke));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_STORE_DESTROY, reinterpret_cast<void*>(JniCacheStoreDestroy));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_STORE_SESSION_CREATE, reinterpret_cast<void*>(JniCacheStoreSessionCreate));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_ENTRY_FILTER_CREATE, reinterpret_cast<void*>(JniCacheEntryFilterCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_ENTRY_FILTER_APPLY, reinterpret_cast<void*>(JniCacheEntryFilterApply));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_ENTRY_FILTER_DESTROY, reinterpret_cast<void*>(JniCacheEntryFilterDestroy));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CACHE_INVOKE, reinterpret_cast<void*>(JniCacheInvoke));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_TASK_MAP, reinterpret_cast<void*>(JniComputeTaskMap));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_TASK_JOB_RESULT, reinterpret_cast<void*>(JniComputeTaskJobResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_TASK_REDUCE, reinterpret_cast<void*>(JniComputeTaskReduce));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_TASK_COMPLETE, reinterpret_cast<void*>(JniComputeTaskComplete));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_JOB_SERIALIZE, reinterpret_cast<void*>(JniComputeJobSerialize));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_JOB_CREATE, reinterpret_cast<void*>(JniComputeJobCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_JOB_EXECUTE, reinterpret_cast<void*>(JniComputeJobExecute));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_JOB_DESTROY, reinterpret_cast<void*>(JniComputeJobDestroy));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_COMPUTE_JOB_CANCEL, reinterpret_cast<void*>(JniComputeJobCancel));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CONTINUOUS_QUERY_LSNR_APPLY, reinterpret_cast<void*>(JniContinuousQueryListenerApply));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CONTINUOUS_QUERY_FILTER_CREATE, reinterpret_cast<void*>(JniContinuousQueryFilterCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CONTINUOUS_QUERY_FILTER_EVAL, reinterpret_cast<void*>(JniContinuousQueryFilterApply));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CONTINUOUS_QUERY_FILTER_RELEASE, reinterpret_cast<void*>(JniContinuousQueryFilterRelease));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_DATA_STREAMER_TOPOLOGY_UPDATE, reinterpret_cast<void*>(JniDataStreamerTopologyUpdate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_DATA_STREAMER_STREAM_RECEIVER_INVOKE, reinterpret_cast<void*>(JniDataStreamerStreamReceiverInvoke));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_BYTE_RES, reinterpret_cast<void*>(JniFutureByteResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_BOOL_RES, reinterpret_cast<void*>(JniFutureBoolResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_SHORT_RES, reinterpret_cast<void*>(JniFutureShortResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_CHAR_RES, reinterpret_cast<void*>(JniFutureCharResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_INT_RES, reinterpret_cast<void*>(JniFutureIntResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_FLOAT_RES, reinterpret_cast<void*>(JniFutureFloatResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_LONG_RES, reinterpret_cast<void*>(JniFutureLongResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_DOUBLE_RES, reinterpret_cast<void*>(JniFutureDoubleResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_OBJ_RES, reinterpret_cast<void*>(JniFutureObjectResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_NULL_RES, reinterpret_cast<void*>(JniFutureNullResult));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_FUTURE_ERR, reinterpret_cast<void*>(JniFutureError));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_LIFECYCLE_EVENT, reinterpret_cast<void*>(JniLifecycleEvent));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_MESSAGING_FILTER_CREATE, reinterpret_cast<void*>(JniMessagingFilterCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_MESSAGING_FILTER_APPLY, reinterpret_cast<void*>(JniMessagingFilterApply));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_MESSAGING_FILTER_DESTROY, reinterpret_cast<void*>(JniMessagingFilterDestroy));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_EVENT_FILTER_CREATE, reinterpret_cast<void*>(JniEventFilterCreate));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_EVENT_FILTER_APPLY, reinterpret_cast<void*>(JniEventFilterApply));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_EVENT_FILTER_DESTROY, reinterpret_cast<void*>(JniEventFilterDestroy));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_SERVICE_INIT, reinterpret_cast<void*>(JniServiceInit));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_SERVICE_EXECUTE, reinterpret_cast<void*>(JniServiceExecute));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_SERVICE_CANCEL, reinterpret_cast<void*>(JniServiceCancel));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_SERVICE_INVOKE_METHOD, reinterpret_cast<void*>(JniServiceInvokeMethod));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CLUSTER_NODE_FILTER_APPLY, reinterpret_cast<void*>(JniClusterNodeFilterApply));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_NODE_INFO, reinterpret_cast<void*>(JniNodeInfo));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_MEMORY_REALLOCATE, reinterpret_cast<void*>(JniMemoryReallocate));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_ON_START, reinterpret_cast<void*>(JniOnStart));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_ON_STOP, reinterpret_cast<void*>(JniOnStop));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_EXTENSION_CALLBACK_IN_LONG_OUT_LONG, reinterpret_cast<void*>(JniExtensionCallbackInLongOutLong));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_EXTENSION_CALLBACK_IN_LONG_LONG_OUT_LONG, reinterpret_cast<void*>(JniExtensionCallbackInLongLongOutLong));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_ON_CLIENT_DISCONNECTED, reinterpret_cast<void*>(JniOnClientDisconnected));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_ON_CLIENT_RECONNECTED, reinterpret_cast<void*>(JniOnClientReconnected));
-
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_AFFINITY_FUNCTION_INIT, reinterpret_cast<void*>(JniAffinityFunctionInit));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_AFFINITY_FUNCTION_PARTITION, reinterpret_cast<void*>(JniAffinityFunctionPartition));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_AFFINITY_FUNCTION_ASSIGN_PARTITIONS, reinterpret_cast<void*>(JniAffinityFunctionAssignPartitions));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_AFFINITY_FUNCTION_REMOVE_NODE, reinterpret_cast<void*>(JniAffinityFunctionRemoveNode));
- AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_AFFINITY_FUNCTION_DESTROY, reinterpret_cast<void*>(JniAffinityFunctionDestroy));
AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_CONSOLE_WRITE, reinterpret_cast<void*>(JniConsoleWrite));
AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_LOGGER_LOG, reinterpret_cast<void*>(JniLoggerLog));
AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_LOGGER_IS_LEVEL_ENABLED, reinterpret_cast<void*>(JniLoggerIsLevelEnabled));
+ AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_IN_LONG_OUT_LONG, reinterpret_cast<void*>(JniInLongOutLong));
+ AddNativeMethod(methods + idx++, M_PLATFORM_CALLBACK_UTILS_IN_LONG_LONG_LONG_OBJECT_OUT_LONG, reinterpret_cast<void*>(JniInLongLongLongObjectOutLong));
+
jint res = env->RegisterNatives(FindClass(env, C_PLATFORM_CALLBACK_UTILS), methods, idx);
if (res != JNI_OK)
@@ -1617,250 +1546,6 @@ namespace ignite
return NULL;
}
- JNIEXPORT jlong JNICALL JniCacheStoreCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, CacheStoreCreateHandler, cacheStoreCreate, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniCacheStoreInvoke(JNIEnv *env, jclass cls, jlong envPtr, jlong objPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, CacheStoreInvokeHandler, cacheStoreInvoke, objPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniCacheStoreDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong objPtr) {
- IGNITE_SAFE_PROC(env, envPtr, CacheStoreDestroyHandler, cacheStoreDestroy, objPtr);
- }
-
- JNIEXPORT jlong JNICALL JniCacheStoreSessionCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong storePtr) {
- IGNITE_SAFE_FUNC(env, envPtr, CacheStoreSessionCreateHandler, cacheStoreSessionCreate, storePtr);
- }
-
- JNIEXPORT jlong JNICALL JniCacheEntryFilterCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, CacheEntryFilterCreateHandler, cacheEntryFilterCreate, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniCacheEntryFilterApply(JNIEnv *env, jclass cls, jlong envPtr, jlong objPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, CacheEntryFilterApplyHandler, cacheEntryFilterApply, objPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniCacheEntryFilterDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong objPtr) {
- IGNITE_SAFE_PROC(env, envPtr, CacheEntryFilterDestroyHandler, cacheEntryFilterDestroy, objPtr);
- }
-
- JNIEXPORT void JNICALL JniCacheInvoke(JNIEnv *env, jclass cls, jlong envPtr, jlong inMemPtr, jlong outMemPtr) {
- IGNITE_SAFE_PROC(env, envPtr, CacheInvokeHandler, cacheInvoke, inMemPtr, outMemPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeTaskMap(JNIEnv *env, jclass cls, jlong envPtr, jlong taskPtr, jlong inMemPtr, jlong outMemPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeTaskMapHandler, computeTaskMap, taskPtr, inMemPtr, outMemPtr);
- }
-
- JNIEXPORT jint JNICALL JniComputeTaskJobResult(JNIEnv *env, jclass cls, jlong envPtr, jlong taskPtr, jlong jobPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ComputeTaskJobResultHandler, computeTaskJobRes, taskPtr, jobPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeTaskReduce(JNIEnv *env, jclass cls, jlong envPtr, jlong taskPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeTaskReduceHandler, computeTaskReduce, taskPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeTaskComplete(JNIEnv *env, jclass cls, jlong envPtr, jlong taskPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeTaskCompleteHandler, computeTaskComplete, taskPtr, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniComputeJobSerialize(JNIEnv *env, jclass cls, jlong envPtr, jlong jobPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ComputeJobSerializeHandler, computeJobSerialize, jobPtr, memPtr);
- }
-
- JNIEXPORT jlong JNICALL JniComputeJobCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ComputeJobCreateHandler, computeJobCreate, memPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeJobExecute(JNIEnv *env, jclass cls, jlong envPtr, jlong jobPtr, jint cancel, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeJobExecuteHandler, computeJobExec, jobPtr, cancel, memPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeJobCancel(JNIEnv *env, jclass cls, jlong envPtr, jlong jobPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeJobCancelHandler, computeJobCancel, jobPtr);
- }
-
- JNIEXPORT void JNICALL JniComputeJobDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong jobPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ComputeJobDestroyHandler, computeJobDestroy, jobPtr);
- }
-
- JNIEXPORT void JNICALL JniContinuousQueryListenerApply(JNIEnv *env, jclass cls, jlong envPtr, jlong cbPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ContinuousQueryListenerApplyHandler, contQryLsnrApply, cbPtr, memPtr);
- }
-
- JNIEXPORT jlong JNICALL JniContinuousQueryFilterCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ContinuousQueryFilterCreateHandler, contQryFilterCreate, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniContinuousQueryFilterApply(JNIEnv *env, jclass cls, jlong envPtr, jlong filterPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ContinuousQueryFilterApplyHandler, contQryFilterApply, filterPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniContinuousQueryFilterRelease(JNIEnv *env, jclass cls, jlong envPtr, jlong filterPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ContinuousQueryFilterReleaseHandler, contQryFilterRelease, filterPtr);
- }
-
- JNIEXPORT void JNICALL JniDataStreamerTopologyUpdate(JNIEnv *env, jclass cls, jlong envPtr, jlong ldrPtr, jlong topVer, jint topSize) {
- IGNITE_SAFE_PROC(env, envPtr, DataStreamerTopologyUpdateHandler, dataStreamerTopologyUpdate, ldrPtr, topVer, topSize);
- }
-
- JNIEXPORT void JNICALL JniDataStreamerStreamReceiverInvoke(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jobject cache, jlong memPtr, jboolean keepPortable) {
- jobject cache0 = env->NewGlobalRef(cache);
-
- if (cache0)
- {
- JniGlobalRefGuard guard(env, cache0);
-
- IGNITE_SAFE_PROC(env, envPtr, DataStreamerStreamReceiverInvokeHandler, streamReceiverInvoke, ptr, cache0, memPtr, keepPortable);
- }
- }
-
- JNIEXPORT void JNICALL JniFutureByteResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jint res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureByteResultHandler, futByteRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureBoolResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jint res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureBoolResultHandler, futBoolRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureShortResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jint res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureShortResultHandler, futShortRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureCharResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jint res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureCharResultHandler, futCharRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureIntResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jint res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureIntResultHandler, futIntRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureFloatResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jfloat res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureFloatResultHandler, futFloatRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureLongResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jlong res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureLongResultHandler, futLongRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureDoubleResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jdouble res) {
- IGNITE_SAFE_PROC(env, envPtr, FutureDoubleResultHandler, futDoubleRes, futPtr, res);
- }
-
- JNIEXPORT void JNICALL JniFutureObjectResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, FutureObjectResultHandler, futObjRes, futPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniFutureNullResult(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr) {
- IGNITE_SAFE_PROC(env, envPtr, FutureNullResultHandler, futNullRes, futPtr);
- }
-
- JNIEXPORT void JNICALL JniFutureError(JNIEnv *env, jclass cls, jlong envPtr, jlong futPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, FutureErrorHandler, futErr, futPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniLifecycleEvent(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jint evt) {
- IGNITE_SAFE_PROC(env, envPtr, LifecycleEventHandler, lifecycleEvt, ptr, evt);
- }
-
- JNIEXPORT void JNICALL JniMemoryReallocate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr, jint cap) {
- IGNITE_SAFE_PROC(env, envPtr, MemoryReallocateHandler, memRealloc, memPtr, cap);
- }
-
- JNIEXPORT jlong JNICALL JniMessagingFilterCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, MessagingFilterCreateHandler, messagingFilterCreate, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniMessagingFilterApply(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, MessagingFilterApplyHandler, messagingFilterApply, ptr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniMessagingFilterDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr) {
- IGNITE_SAFE_PROC(env, envPtr, MessagingFilterDestroyHandler, messagingFilterDestroy, ptr);
- }
-
- JNIEXPORT jlong JNICALL JniEventFilterCreate(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, EventFilterCreateHandler, eventFilterCreate, memPtr);
- }
-
- JNIEXPORT jint JNICALL JniEventFilterApply(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, EventFilterApplyHandler, eventFilterApply, ptr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniEventFilterDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr) {
- IGNITE_SAFE_PROC(env, envPtr, EventFilterDestroyHandler, eventFilterDestroy, ptr);
- }
-
- JNIEXPORT jlong JNICALL JniServiceInit(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ServiceInitHandler, serviceInit, memPtr);
- }
-
- JNIEXPORT void JNICALL JniServiceExecute(JNIEnv *env, jclass cls, jlong envPtr, jlong svcPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ServiceExecuteHandler, serviceExecute, svcPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniServiceCancel(JNIEnv *env, jclass cls, jlong envPtr, jlong svcPtr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ServiceCancelHandler, serviceCancel, svcPtr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniServiceInvokeMethod(JNIEnv *env, jclass cls, jlong envPtr, jlong svcPtr, jlong inMemPtr, jlong outMemPtr) {
- IGNITE_SAFE_PROC(env, envPtr, ServiceInvokeMethodHandler, serviceInvokeMethod, svcPtr, inMemPtr, outMemPtr);
- }
-
- JNIEXPORT jint JNICALL JniClusterNodeFilterApply(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, ClusterNodeFilterApplyHandler, clusterNodeFilterApply, memPtr);
- }
-
- JNIEXPORT jlong JNICALL JniNodeInfo(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, NodeInfoHandler, nodeInfo, memPtr);
- }
-
- JNIEXPORT void JNICALL JniOnStart(JNIEnv *env, jclass cls, jlong envPtr, jobject proc, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, OnStartHandler, onStart, proc, memPtr);
- }
-
- JNIEXPORT void JNICALL JniOnStop(JNIEnv *env, jclass cls, jlong envPtr) {
- IGNITE_SAFE_PROC_NO_ARG(env, envPtr, OnStopHandler, onStop);
- }
-
- JNIEXPORT jlong JNICALL JniExtensionCallbackInLongOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint typ, jlong arg1) {
- IGNITE_SAFE_FUNC(env, envPtr, ExtensionCallbackInLongOutLongHandler, extensionCallbackInLongOutLong, typ, arg1);
- }
-
- JNIEXPORT jlong JNICALL JniExtensionCallbackInLongLongOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint typ, jlong arg1, jlong arg2) {
- IGNITE_SAFE_FUNC(env, envPtr, ExtensionCallbackInLongLongOutLongHandler, extensionCallbackInLongLongOutLong, typ, arg1, arg2);
- }
-
- JNIEXPORT void JNICALL JniOnClientDisconnected(JNIEnv *env, jclass cls, jlong envPtr) {
- IGNITE_SAFE_PROC_NO_ARG(env, envPtr, OnClientDisconnectedHandler, onClientDisconnected);
- }
-
- JNIEXPORT void JNICALL JniOnClientReconnected(JNIEnv *env, jclass cls, jlong envPtr, jboolean clusterRestarted) {
- IGNITE_SAFE_PROC(env, envPtr, OnClientReconnectedHandler, onClientReconnected, clusterRestarted);
- }
-
- JNIEXPORT jlong JNICALL JniAffinityFunctionInit(JNIEnv *env, jclass cls, jlong envPtr, jlong memPtr, jobject baseFunc) {
- void* baseFuncRef = baseFunc ? env->NewGlobalRef(baseFunc) : 0;
- IGNITE_SAFE_FUNC(env, envPtr, AffinityFunctionInitHandler, affinityFunctionInit, memPtr, baseFuncRef);
- }
-
- JNIEXPORT jint JNICALL JniAffinityFunctionPartition(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jlong memPtr) {
- IGNITE_SAFE_FUNC(env, envPtr, AffinityFunctionPartitionHandler, affinityFunctionPartition, ptr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniAffinityFunctionAssignPartitions(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jlong inMemPtr, jlong outMemPtr) {
- IGNITE_SAFE_PROC(env, envPtr, AffinityFunctionAssignPartitionsHandler, affinityFunctionAssignPartitions, ptr, inMemPtr, outMemPtr);
- }
-
- JNIEXPORT void JNICALL JniAffinityFunctionRemoveNode(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr, jlong memPtr) {
- IGNITE_SAFE_PROC(env, envPtr, AffinityFunctionRemoveNodeHandler, affinityFunctionRemoveNode, ptr, memPtr);
- }
-
- JNIEXPORT void JNICALL JniAffinityFunctionDestroy(JNIEnv *env, jclass cls, jlong envPtr, jlong ptr) {
- IGNITE_SAFE_PROC(env, envPtr, AffinityFunctionDestroyHandler, affinityFunctionDestroy, ptr);
- }
-
JNIEXPORT void JNICALL JniConsoleWrite(JNIEnv *env, jclass cls, jstring str, jboolean isErr) {
CONSOLE_LOCK.Enter();
@@ -1903,6 +1588,14 @@ namespace ignite
JNIEXPORT jboolean JNICALL JniLoggerIsLevelEnabled(JNIEnv *env, jclass cls, jlong envPtr, jint level) {
IGNITE_SAFE_FUNC(env, envPtr, LoggerIsLevelEnabledHandler, loggerIsLevelEnabled, level);
}
+
+ JNIEXPORT jlong JNICALL JniInLongOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint type, jlong val) {
+ IGNITE_SAFE_FUNC(env, envPtr, InLongOutLongHandler, inLongOutLong, type, val);
+ }
+
+ JNIEXPORT jlong JNICALL JniInLongLongLongObjectOutLong(JNIEnv *env, jclass cls, jlong envPtr, jint type, jlong val1, jlong val2, jlong val3, jobject arg) {
+ IGNITE_SAFE_FUNC(env, envPtr, InLongLongLongObjectOutLongHandler, inLongLongLongObjectOutLong, type, val1, val2, val3, arg);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
index dcddfdb..c4d4279 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Services/ServicesTest.cs
@@ -438,7 +438,8 @@ namespace Apache.Ignite.Core.Tests.Services
var ex = Assert.Throws<IgniteException>(() => Services.DeployMultiple(SvcName, svc, Grids.Length, 1));
Assert.AreEqual("Expected exception", ex.Message);
Assert.IsNotNull(ex.InnerException);
- Assert.IsTrue(ex.InnerException.Message.Contains("PlatformCallbackUtils.serviceInit(Native Method)"));
+ Assert.IsTrue(ex.InnerException.Message.Contains("PlatformCallbackGateway.serviceInit"),
+ ex.InnerException.Message);
var svc0 = Services.GetService<TestIgniteServiceSerializable>(SvcName);
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
index 6421b8c..b2db0a7 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
@@ -429,6 +429,7 @@
<Compile Include="Impl\Unmanaged\IgniteJniNativeMethods.cs" />
<Compile Include="Impl\Unmanaged\IUnmanagedTarget.cs" />
<Compile Include="Impl\Unmanaged\UnmanagedCallbackHandlers.cs" />
+ <Compile Include="Impl\Unmanaged\UnmanagedCallbackOp.cs" />
<Compile Include="Impl\Unmanaged\UnmanagedCallbacks.cs" />
<Compile Include="Impl\Unmanaged\UnmanagedContext.cs" />
<Compile Include="Impl\Unmanaged\UnmanagedNonReleaseableTarget.cs" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryUtils.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryUtils.cs
index cc5d8a1..e6a3716 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryUtils.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryUtils.cs
@@ -1961,6 +1961,22 @@ namespace Apache.Ignite.Core.Impl.Binary
}
/// <summary>
+ /// Reinterprets int bits as a float.
+ /// </summary>
+ public static unsafe float IntToFloatBits(int val)
+ {
+ return *(float*) &val;
+ }
+
+ /// <summary>
+ /// Reinterprets long bits as a double.
+ /// </summary>
+ public static unsafe double LongToDoubleBits(long val)
+ {
+ return *(double*) &val;
+ }
+
+ /// <summary>
/// Creates and instance from the type name in reader.
/// </summary>
private static T CreateInstance<T>(BinaryReader reader)
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/BinaryStreamBase.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/BinaryStreamBase.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/BinaryStreamBase.cs
index 6286602..184209f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/BinaryStreamBase.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Io/BinaryStreamBase.cs
@@ -579,7 +579,7 @@ namespace Apache.Ignite.Core.Impl.Binary.IO
{
int val = ReadInt();
- return *(float*)(&val);
+ return BinaryUtils.IntToFloatBits(val);
}
/// <summary>
@@ -849,7 +849,7 @@ namespace Apache.Ignite.Core.Impl.Binary.IO
{
long val = ReadLong();
- return *(double*)(&val);
+ return BinaryUtils.LongToDoubleBits(val);
}
/// <summary>
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeTaskHolder.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeTaskHolder.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeTaskHolder.cs
index bb8289b..a6d7cb4 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeTaskHolder.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeTaskHolder.cs
@@ -41,10 +41,9 @@ namespace Apache.Ignite.Core.Impl.Compute
/// <summary>
/// Perform map step.
/// </summary>
- /// <param name="inStream">Stream with IN data (topology info).</param>
- /// <param name="outStream">Stream for OUT data (map result).</param>
+ /// <param name="stream">Stream with IN data (topology info) and for OUT data (map result).</param>
/// <returns>Map with produced jobs.</returns>
- void Map(PlatformMemoryStream inStream, PlatformMemoryStream outStream);
+ void Map(PlatformMemoryStream stream);
/// <summary>
/// Process local job result.
@@ -141,7 +140,7 @@ namespace Apache.Ignite.Core.Impl.Compute
/** <inheritDoc /> */
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes",
Justification = "User code can throw any exception")]
- public void Map(PlatformMemoryStream inStream, PlatformMemoryStream outStream)
+ public void Map(PlatformMemoryStream stream)
{
IList<IClusterNode> subgrid;
@@ -150,7 +149,7 @@ namespace Apache.Ignite.Core.Impl.Compute
var ignite = (Ignite) prj.Ignite;
// 1. Unmarshal topology info if topology changed.
- var reader = prj.Marshaller.StartUnmarshal(inStream);
+ var reader = prj.Marshaller.StartUnmarshal(stream);
if (reader.ReadBoolean())
{
@@ -208,7 +207,8 @@ namespace Apache.Ignite.Core.Impl.Compute
}
// 3. Write map result to the output stream.
- BinaryWriter writer = prj.Marshaller.StartMarshal(outStream);
+ stream.Reset();
+ BinaryWriter writer = prj.Marshaller.StartMarshal(stream);
try
{
@@ -240,7 +240,7 @@ namespace Apache.Ignite.Core.Impl.Compute
// Something went wrong during marshaling.
Finish(default(TR), e);
- outStream.Reset();
+ stream.Reset();
writer.WriteBoolean(false); // Map failed.
writer.WriteString(e.Message); // Write error message.
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackHandlers.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackHandlers.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackHandlers.cs
index 6367e1e..32e0a29 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackHandlers.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackHandlers.cs
@@ -27,85 +27,12 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
{
internal void* target;
- internal void* cacheStoreCreate;
- internal void* cacheStoreInvoke;
- internal void* cacheStoreDestroy;
- internal void* cacheStoreSessionCreate;
-
- internal void* cacheEntryFilterCreate;
- internal void* cacheEntryFilterApply;
- internal void* cacheEntryFilterDestroy;
-
- internal void* cacheInvoke;
-
- internal void* computeTaskMap;
- internal void* computeTaskJobResult;
- internal void* computeTaskReduce;
- internal void* computeTaskComplete;
- internal void* computeJobSerialize;
- internal void* computeJobCreate;
- internal void* computeJobExecute;
- internal void* computeJobCancel;
- internal void* computeJobDestroy;
-
- internal void* continuousQueryListenerApply;
- internal void* continuousQueryFilterCreate;
- internal void* continuousQueryFilterApply;
- internal void* continuousQueryFilterRelease;
-
- internal void* dataStreamerTopologyUpdate;
- internal void* dataStreamerStreamReceiverInvoke;
-
- internal void* futureByteResult;
- internal void* futureBoolResult;
- internal void* futureShortResult;
- internal void* futureCharResult;
- internal void* futureIntResult;
- internal void* futureFloatResult;
- internal void* futureLongResult;
- internal void* futureDoubleResult;
- internal void* futureObjectResult;
- internal void* futureNullResult;
- internal void* futureError;
-
- internal void* lifecycleOnEvent;
-
- internal void* memoryReallocate;
-
- internal void* messagingFilterCreate;
- internal void* messagingFilterApply;
- internal void* messagingFilterDestroy;
-
- internal void* eventFilterCreate;
- internal void* eventFilterApply;
- internal void* eventFilterDestroy;
-
- internal void* serviceInit;
- internal void* serviceExecute;
- internal void* serviceCancel;
- internal void* serviceInvokeMethod;
-
- internal void* clusterNodeFilterApply;
-
- internal void* nodeInfo;
-
- internal void* onStart;
- internal void* onStop;
internal void* error;
- internal void* extensionCbInLongOutLong;
- internal void* extensionCbInLongLongOutLong;
-
- internal void* onClientDisconnected;
- internal void* ocClientReconnected;
-
- internal void* affinityFunctionInit;
- internal void* affinityFunctionPartition;
- internal void* affinityFunctionAssignPartitions;
- internal void* affinityFunctionRemoveNode;
- internal void* affinityFunctionDestroy;
-
internal void* loggerLog;
internal void* loggerIsLevelEnabled;
+
+ internal void* inLongOutLong;
+ internal void* inLongLongObjectOutLong;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackOp.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackOp.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackOp.cs
new file mode 100644
index 0000000..0b766de
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedCallbackOp.cs
@@ -0,0 +1,86 @@
+\ufeff/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Impl.Unmanaged
+{
+ /// <summary>
+ /// Callback op codes.
+ /// </summary>
+ internal enum UnmanagedCallbackOp
+ {
+ CacheStoreCreate = 1,
+ CacheStoreInvoke = 2,
+ CacheStoreDestroy = 3,
+ CacheStoreSessionCreate = 4,
+ CacheEntryFilterCreate = 5,
+ CacheEntryFilterApply = 6,
+ CacheEntryFilterDestroy = 7,
+ CacheInvoke = 8,
+ ComputeTaskMap = 9,
+ ComputeTaskJobResult = 10,
+ ComputeTaskReduce = 11,
+ ComputeTaskComplete = 12,
+ ComputeJobSerialize = 13,
+ ComputeJobCreate = 14,
+ ComputeJobExecute = 15,
+ ComputeJobCancel = 16,
+ ComputeJobDestroy = 17,
+ ContinuousQueryListenerApply = 18,
+ ContinuousQueryFilterCreate = 19,
+ ContinuousQueryFilterApply = 20,
+ ContinuousQueryFilterRelease = 21,
+ DataStreamerTopologyUpdate = 22,
+ DataStreamerStreamReceiverInvoke = 23,
+ FutureByteResult = 24,
+ FutureBoolResult = 25,
+ FutureShortResult = 26,
+ FutureCharResult = 27,
+ FutureIntResult = 28,
+ FutureFloatResult = 29,
+ FutureLongResult = 30,
+ FutureDoubleResult = 31,
+ FutureObjectResult = 32,
+ FutureNullResult = 33,
+ FutureError = 34,
+ LifecycleOnEvent = 35,
+ MemoryReallocate = 36,
+ MessagingFilterCreate = 37,
+ MessagingFilterApply = 38,
+ MessagingFilterDestroy = 39,
+ EventFilterCreate = 40,
+ EventFilterApply = 41,
+ EventFilterDestroy = 42,
+ ServiceInit = 43,
+ ServiceExecute = 44,
+ ServiceCancel = 45,
+ ServiceInvokeMethod = 46,
+ ClusterNodeFilterApply = 47,
+ NodeInfo = 48,
+ OnStart = 49,
+ OnStop = 50,
+ ExtensionInLongLongOutLong = 52,
+ OnClientDisconnected = 53,
+ OnClientReconnected = 54,
+ AffinityFunctionInit = 55,
+ AffinityFunctionPartition = 56,
+ AffinityFunctionAssignPartitions = 57,
+ AffinityFunctionRemoveNode = 58,
+ AffinityFunctionDestroy = 59,
+ ComputeTaskLocalJobResult = 60,
+ ComputeJobExecuteLocal = 61
+ }
+}
[3/3] ignite git commit: IGNITE-4033 Streamline platform callback
interface
Posted by pt...@apache.org.
IGNITE-4033 Streamline platform callback interface
This closes #1261
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/72ac53da
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/72ac53da
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/72ac53da
Branch: refs/heads/master
Commit: 72ac53da2e6f8311c2d816763a6765724b79491a
Parents: f621f7f
Author: Pavel Tupitsyn <pt...@apache.org>
Authored: Mon Dec 12 17:44:48 2016 +0300
Committer: Pavel Tupitsyn <pt...@apache.org>
Committed: Mon Dec 12 17:44:48 2016 +0300
----------------------------------------------------------------------
.../cache/PlatformCacheEntryFilterImpl.java | 4 +-
.../cache/PlatformCacheEntryProcessorImpl.java | 43 +-
.../affinity/PlatformAffinityFunction.java | 52 +-
.../callback/PlatformCallbackGateway.java | 266 ++--
.../platform/callback/PlatformCallbackOp.java | 206 +++
.../callback/PlatformCallbackUtils.java | 544 +-------
.../platform/compute/PlatformAbstractJob.java | 2 +-
.../platform/compute/PlatformAbstractTask.java | 15 +-
.../platform/compute/PlatformClosureJob.java | 12 +-
.../platform/compute/PlatformFullJob.java | 15 +-
.../platform/compute/PlatformFullTask.java | 18 +-
.../PlatformStreamReceiverImpl.java | 3 +
.../dotnet/PlatformDotNetCacheStore.java | 6 +-
.../services/PlatformAbstractService.java | 25 +-
.../platform/utils/PlatformFutureUtils.java | 4 +-
.../platform/utils/PlatformUtils.java | 8 +-
.../cpp/core/src/impl/ignite_environment.cpp | 74 +-
.../platforms/cpp/jni/include/ignite/jni/java.h | 85 +-
modules/platforms/cpp/jni/project/vs/module.def | 4 +-
modules/platforms/cpp/jni/src/java.cpp | 339 +----
.../Services/ServicesTest.cs | 3 +-
.../Apache.Ignite.Core.csproj | 1 +
.../Impl/Binary/BinaryUtils.cs | 16 +
.../Impl/Binary/Io/BinaryStreamBase.cs | 4 +-
.../Impl/Compute/ComputeTaskHolder.cs | 14 +-
.../Impl/Unmanaged/UnmanagedCallbackHandlers.cs | 79 +-
.../Impl/Unmanaged/UnmanagedCallbackOp.cs | 86 ++
.../Impl/Unmanaged/UnmanagedCallbacks.cs | 1229 +++++++++---------
28 files changed, 1293 insertions(+), 1864 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilterImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilterImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilterImpl.java
index 4c86d6d..3c55b76 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilterImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilterImpl.java
@@ -60,12 +60,14 @@ public class PlatformCacheEntryFilterImpl extends PlatformAbstractPredicate impl
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(ptr);
+
writer.writeObject(k);
writer.writeObject(v);
out.synchronize();
- return ctx.gateway().cacheEntryFilterApply(ptr, mem.pointer()) != 0;
+ return ctx.gateway().cacheEntryFilterApply(mem.pointer()) != 0;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryProcessorImpl.java
index 3e8ad61..31dd267 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryProcessorImpl.java
@@ -17,15 +17,9 @@
package org.apache.ignite.internal.processors.platform.cache;
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import javax.cache.processor.EntryProcessorException;
-import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.binary.BinaryRawWriter;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformContext;
@@ -36,6 +30,13 @@ import org.apache.ignite.internal.processors.platform.memory.PlatformOutputStrea
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
import org.apache.ignite.internal.util.typedef.internal.U;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
/**
* Platform cache entry processor. Delegates processing to native platform.
*/
@@ -65,7 +66,7 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
private transient long ptr;
/**
- * {@link java.io.Externalizable} support.
+ * {@link Externalizable} support.
*/
public PlatformCacheEntryProcessorImpl() {
// No-op.
@@ -86,7 +87,7 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
@Override public Object process(MutableEntry entry, Object... args)
throws EntryProcessorException {
try {
- IgniteKernal ignite = (IgniteKernal)entry.unwrap(Ignite.class);
+ Ignite ignite = (Ignite)entry.unwrap(Ignite.class);
PlatformProcessor interopProc;
@@ -112,12 +113,10 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
* @param ctx Context.
* @param entry Entry.
* @return Processing result.
- * @throws org.apache.ignite.IgniteCheckedException
*/
- private Object execute0(PlatformContext ctx, MutableEntry entry)
- throws IgniteCheckedException {
- try (PlatformMemory outMem = ctx.memory().allocate()) {
- PlatformOutputStream out = outMem.output();
+ private Object execute0(PlatformContext ctx, MutableEntry entry) {
+ try (PlatformMemory mem = ctx.memory().allocate()) {
+ PlatformOutputStream out = mem.output();
BinaryRawWriterEx writer = ctx.writer(out);
@@ -125,17 +124,15 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
out.synchronize();
- try (PlatformMemory inMem = ctx.memory().allocate()) {
- PlatformInputStream in = inMem.input();
+ ctx.gateway().cacheInvoke(mem.pointer());
- ctx.gateway().cacheInvoke(outMem.pointer(), inMem.pointer());
+ PlatformInputStream in = mem.input();
- in.synchronize();
+ in.synchronize();
- BinaryRawReaderEx reader = ctx.reader(in);
+ BinaryRawReaderEx reader = ctx.reader(in);
- return readResultAndUpdateEntry(ctx, entry, reader);
- }
+ return readResultAndUpdateEntry(ctx, entry, reader);
}
}
@@ -145,7 +142,7 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
* @param entry Entry to process.
* @param writer Writer.
*/
- private void writeEntryAndProcessor(MutableEntry entry, BinaryRawWriterEx writer) {
+ private void writeEntryAndProcessor(MutableEntry entry, BinaryRawWriter writer) {
writer.writeObject(entry.getKey());
writer.writeObject(entry.getValue());
@@ -167,7 +164,7 @@ public class PlatformCacheEntryProcessorImpl implements PlatformCacheEntryProces
* @param entry Mutable entry to update.
* @param reader Reader.
* @return Entry processing result
- * @throws javax.cache.processor.EntryProcessorException If processing has failed in user code.
+ * @throws EntryProcessorException If processing has failed in user code.
*/
@SuppressWarnings("unchecked")
private Object readResultAndUpdateEntry(PlatformContext ctx, MutableEntry entry, BinaryRawReaderEx reader) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/affinity/PlatformAffinityFunction.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/affinity/PlatformAffinityFunction.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/affinity/PlatformAffinityFunction.java
index 2d3cada..4206d40 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/affinity/PlatformAffinityFunction.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/affinity/PlatformAffinityFunction.java
@@ -166,11 +166,12 @@ public class PlatformAffinityFunction implements AffinityFunction, Externalizabl
PlatformOutputStream out = mem.output();
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(ptr);
writer.writeObject(key);
out.synchronize();
- return ctx.gateway().affinityFunctionPartition(ptr, mem.pointer());
+ return ctx.gateway().affinityFunctionPartition(mem.pointer());
}
}
@@ -186,34 +187,34 @@ public class PlatformAffinityFunction implements AffinityFunction, Externalizabl
assert ptr != 0;
assert affCtx != null;
- try (PlatformMemory outMem = ctx.memory().allocate()) {
- try (PlatformMemory inMem = ctx.memory().allocate()) {
- PlatformOutputStream out = outMem.output();
- BinaryRawWriterEx writer = ctx.writer(out);
+ try (PlatformMemory mem = ctx.memory().allocate()) {
+ PlatformOutputStream out = mem.output();
+ BinaryRawWriterEx writer = ctx.writer(out);
+
+ writer.writeLong(ptr);
- // Write previous assignment
- PlatformAffinityUtils.writeAffinityFunctionContext(affCtx, writer, ctx);
+ // Write previous assignment
+ PlatformAffinityUtils.writeAffinityFunctionContext(affCtx, writer, ctx);
- out.synchronize();
+ out.synchronize();
+
+ // Call platform
+ // We can not restore original AffinityFunctionContext after the call to platform,
+ // due to DiscoveryEvent (when node leaves, we can't get it by id anymore).
+ // Secondly, AffinityFunctionContext can't be changed by the user.
+ if (baseTarget != null)
+ baseTarget.setCurrentAffinityFunctionContext(affCtx);
- // Call platform
- // We can not restore original AffinityFunctionContext after the call to platform,
- // due to DiscoveryEvent (when node leaves, we can't get it by id anymore).
- // Secondly, AffinityFunctionContext can't be changed by the user.
+ try {
+ ctx.gateway().affinityFunctionAssignPartitions(mem.pointer());
+ }
+ finally {
if (baseTarget != null)
- baseTarget.setCurrentAffinityFunctionContext(affCtx);
-
- try {
- ctx.gateway().affinityFunctionAssignPartitions(ptr, outMem.pointer(), inMem.pointer());
- }
- finally {
- if (baseTarget != null)
- baseTarget.setCurrentAffinityFunctionContext(null);
- }
-
- // Read result
- return PlatformAffinityUtils.readPartitionAssignment(ctx.reader(inMem), ctx);
+ baseTarget.setCurrentAffinityFunctionContext(null);
}
+
+ // Read result
+ return PlatformAffinityUtils.readPartitionAssignment(ctx.reader(mem), ctx);
}
}
@@ -234,11 +235,12 @@ public class PlatformAffinityFunction implements AffinityFunction, Externalizabl
PlatformOutputStream out = mem.output();
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(ptr);
writer.writeUuid(nodeId);
out.synchronize();
- ctx.gateway().affinityFunctionRemoveNode(ptr, mem.pointer());
+ ctx.gateway().affinityFunctionRemoveNode(mem.pointer());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackGateway.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackGateway.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackGateway.java
index c77f501..a9268fe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackGateway.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackGateway.java
@@ -61,7 +61,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.cacheStoreCreate(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheStoreCreate, memPtr);
}
finally {
leave();
@@ -69,15 +69,14 @@ public class PlatformCallbackGateway {
}
/**
- * @param objPtr Object pointer.
* @param memPtr Memory pointer.
* @return Result.
*/
- public int cacheStoreInvoke(long objPtr, long memPtr) {
+ public int cacheStoreInvoke(long memPtr) {
enter();
try {
- return PlatformCallbackUtils.cacheStoreInvoke(envPtr, objPtr, memPtr);
+ return (int)PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheStoreInvoke, memPtr);
}
finally {
leave();
@@ -92,7 +91,7 @@ public class PlatformCallbackGateway {
return; // no need to destroy stores on grid stop
try {
- PlatformCallbackUtils.cacheStoreDestroy(envPtr, objPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheStoreDestroy, objPtr);
}
finally {
leave();
@@ -102,14 +101,13 @@ public class PlatformCallbackGateway {
/**
* Creates cache store session.
*
- * @param storePtr Store instance pointer.
* @return Session instance pointer.
*/
- public long cacheStoreSessionCreate(long storePtr) {
+ public long cacheStoreSessionCreate() {
enter();
try {
- return PlatformCallbackUtils.cacheStoreSessionCreate(envPtr, storePtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheStoreSessionCreate, 0);
}
finally {
leave();
@@ -126,7 +124,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.cacheEntryFilterCreate(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheEntryFilterCreate, memPtr);
}
finally {
leave();
@@ -134,15 +132,14 @@ public class PlatformCallbackGateway {
}
/**
- * @param ptr Pointer.
* @param memPtr Memory pointer.
* @return Result.
*/
- public int cacheEntryFilterApply(long ptr, long memPtr) {
+ public int cacheEntryFilterApply(long memPtr) {
enter();
try {
- return PlatformCallbackUtils.cacheEntryFilterApply(envPtr, ptr, memPtr);
+ return (int)PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheEntryFilterApply, memPtr);
}
finally {
leave();
@@ -156,7 +153,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.cacheEntryFilterDestroy(envPtr, ptr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheEntryFilterDestroy, ptr);
}
finally {
leave();
@@ -166,14 +163,13 @@ public class PlatformCallbackGateway {
/**
* Invoke cache entry processor.
*
- * @param outMemPtr Output memory pointer.
- * @param inMemPtr Input memory pointer.
+ * @param memPtr Memory pointer.
*/
- public void cacheInvoke(long outMemPtr, long inMemPtr) {
+ public void cacheInvoke(long memPtr) {
enter();
try {
- PlatformCallbackUtils.cacheInvoke(envPtr, outMemPtr, inMemPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.CacheInvoke, memPtr);
}
finally {
leave();
@@ -183,15 +179,13 @@ public class PlatformCallbackGateway {
/**
* Perform native task map. Do not throw exceptions, serializing them to the output stream instead.
*
- * @param taskPtr Task pointer.
- * @param outMemPtr Output memory pointer (exists if topology changed, otherwise {@code 0}).
- * @param inMemPtr Input memory pointer.
+ * @param memPtr Memory pointer.
*/
- public void computeTaskMap(long taskPtr, long outMemPtr, long inMemPtr) {
+ public void computeTaskMap(long memPtr) {
enter();
try {
- PlatformCallbackUtils.computeTaskMap(envPtr, taskPtr, outMemPtr, inMemPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeTaskMap, memPtr);
}
finally {
leave();
@@ -203,14 +197,31 @@ public class PlatformCallbackGateway {
*
* @param taskPtr Task pointer.
* @param jobPtr Job pointer.
- * @param memPtr Memory pointer (always zero for local job execution).
* @return Job result enum ordinal.
*/
- public int computeTaskJobResult(long taskPtr, long jobPtr, long memPtr) {
+ public int computeTaskLocalJobResult(long taskPtr, long jobPtr) {
+ enter();
+
+ try {
+ return (int)PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ComputeTaskLocalJobResult, taskPtr, jobPtr, 0, null);
+ }
+ finally {
+ leave();
+ }
+ }
+
+ /**
+ * Perform native task job result notification.
+ *
+ * @param memPtr Memory pointer.
+ * @return Job result enum ordinal.
+ */
+ public int computeTaskJobResult(long memPtr) {
enter();
try {
- return PlatformCallbackUtils.computeTaskJobResult(envPtr, taskPtr, jobPtr, memPtr);
+ return (int)PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeTaskJobResult, memPtr);
}
finally {
leave();
@@ -226,7 +237,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.computeTaskReduce(envPtr, taskPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeTaskReduce, taskPtr);
}
finally {
leave();
@@ -243,7 +254,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.computeTaskComplete(envPtr, taskPtr, memPtr);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ComputeTaskComplete, taskPtr, memPtr, 0, null);
}
finally {
leave();
@@ -261,7 +273,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.computeJobSerialize(envPtr, jobPtr, memPtr);
+ return (int)PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ComputeJobSerialize, jobPtr, memPtr, 0, null);
}
finally {
leave();
@@ -278,7 +291,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.computeJobCreate(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeJobCreate, memPtr);
}
finally {
leave();
@@ -290,13 +303,29 @@ public class PlatformCallbackGateway {
*
* @param jobPtr Job pointer.
* @param cancel Cancel flag.
- * @param memPtr Memory pointer to write result to for remote job execution or {@code 0} for local job execution.
*/
- public void computeJobExecute(long jobPtr, int cancel, long memPtr) {
+ public void computeJobExecuteLocal(long jobPtr, long cancel) {
enter();
try {
- PlatformCallbackUtils.computeJobExecute(envPtr, jobPtr, cancel, memPtr);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ComputeJobExecuteLocal, jobPtr, cancel, 0, null);
+ }
+ finally {
+ leave();
+ }
+ }
+
+ /**
+ * Execute native job on a node other than where it was created.
+ *
+ * @param memPtr Memory pointer.
+ */
+ public void computeJobExecute(long memPtr) {
+ enter();
+
+ try {
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeJobExecute, memPtr);
}
finally {
leave();
@@ -312,7 +341,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.computeJobCancel(envPtr, jobPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeJobCancel, jobPtr);
}
finally {
leave();
@@ -328,7 +357,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.computeJobDestroy(envPtr, ptr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ComputeJobDestroy, ptr);
}
finally {
leave();
@@ -338,14 +367,13 @@ public class PlatformCallbackGateway {
/**
* Invoke local callback.
*
- * @param cbPtr Callback pointer.
* @param memPtr Memory pointer.
*/
- public void continuousQueryListenerApply(long cbPtr, long memPtr) {
+ public void continuousQueryListenerApply(long memPtr) {
enter();
try {
- PlatformCallbackUtils.continuousQueryListenerApply(envPtr, cbPtr, memPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ContinuousQueryListenerApply, memPtr);
}
finally {
leave();
@@ -362,7 +390,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.continuousQueryFilterCreate(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ContinuousQueryFilterCreate, memPtr);
}
finally {
leave();
@@ -372,15 +400,14 @@ public class PlatformCallbackGateway {
/**
* Invoke remote filter.
*
- * @param filterPtr Filter pointer.
* @param memPtr Memory pointer.
* @return Result.
*/
- public int continuousQueryFilterApply(long filterPtr, long memPtr) {
+ public long continuousQueryFilterApply(long memPtr) {
enter();
try {
- return PlatformCallbackUtils.continuousQueryFilterApply(envPtr, filterPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ContinuousQueryFilterApply, memPtr);
}
finally {
leave();
@@ -396,7 +423,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.continuousQueryFilterRelease(envPtr, filterPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr,
+ PlatformCallbackOp.ContinuousQueryFilterRelease, filterPtr);
}
finally {
leave();
@@ -414,7 +442,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.dataStreamerTopologyUpdate(envPtr, ptr, topVer, topSize);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.DataStreamerTopologyUpdate, ptr, topVer, topSize, null);
}
finally {
leave();
@@ -424,16 +453,15 @@ public class PlatformCallbackGateway {
/**
* Invoke stream receiver.
*
- * @param ptr Receiver native pointer.
* @param cache Cache object.
* @param memPtr Stream pointer.
- * @param keepBinary Binary flag.
*/
public void dataStreamerStreamReceiverInvoke(long ptr, PlatformTargetProxy cache, long memPtr, boolean keepBinary) {
enter();
try {
- PlatformCallbackUtils.dataStreamerStreamReceiverInvoke(envPtr, ptr, cache, memPtr, keepBinary);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.DataStreamerStreamReceiverInvoke, memPtr, 0, 0, cache);
}
finally {
leave();
@@ -446,11 +474,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureByteResult(long futPtr, int res) {
+ public void futureByteResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureByteResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureByteResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -463,11 +492,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureBoolResult(long futPtr, int res) {
+ public void futureBoolResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureBoolResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureBoolResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -480,11 +510,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureShortResult(long futPtr, int res) {
+ public void futureShortResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureShortResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureShortResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -497,11 +528,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureCharResult(long futPtr, int res) {
+ public void futureCharResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureCharResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureCharResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -514,11 +546,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureIntResult(long futPtr, int res) {
+ public void futureIntResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureIntResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureIntResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -531,11 +564,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureFloatResult(long futPtr, float res) {
+ public void futureFloatResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureFloatResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureFloatResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -552,7 +586,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.futureLongResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureLongResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -565,11 +600,12 @@ public class PlatformCallbackGateway {
* @param futPtr Future pointer.
* @param res Result.
*/
- public void futureDoubleResult(long futPtr, double res) {
+ public void futureDoubleResult(long futPtr, long res) {
enter();
try {
- PlatformCallbackUtils.futureDoubleResult(envPtr, futPtr, res);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureDoubleResult, futPtr, res, 0, null);
}
finally {
leave();
@@ -586,7 +622,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.futureObjectResult(envPtr, futPtr, memPtr);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureObjectResult, futPtr, memPtr, 0, null);
}
finally {
leave();
@@ -602,7 +639,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.futureNullResult(envPtr, futPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.FutureNullResult, futPtr);
}
finally {
leave();
@@ -619,7 +656,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.futureError(envPtr, futPtr, memPtr);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.FutureError, futPtr, memPtr, 0, null);
}
finally {
leave();
@@ -636,8 +674,9 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.messagingFilterCreate(envPtr, memPtr);
- }
+ return PlatformCallbackUtils.inLongOutLong(envPtr,
+ PlatformCallbackOp.MessagingFilterCreate, memPtr);
+ }
finally {
leave();
}
@@ -652,7 +691,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.messagingFilterApply(envPtr, ptr, memPtr);
+ return (int)PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.MessagingFilterApply, ptr, memPtr, 0, null);
}
finally {
leave();
@@ -665,7 +705,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.messagingFilterDestroy(envPtr, ptr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.MessagingFilterDestroy, ptr);
}
finally {
leave();
@@ -682,7 +722,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.eventFilterCreate(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.EventFilterCreate, memPtr);
}
finally {
leave();
@@ -698,7 +738,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.eventFilterApply(envPtr, ptr, memPtr);
+ return (int)PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.EventFilterApply, ptr, memPtr, 0, null);
}
finally {
leave();
@@ -712,7 +753,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.eventFilterDestroy(envPtr, ptr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.EventFilterDestroy, ptr);
}
finally {
leave();
@@ -728,7 +769,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.nodeInfo(envPtr, memPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.NodeInfo, memPtr);
}
finally {
leave();
@@ -745,7 +786,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.onStart(envPtr, proc, memPtr);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr, PlatformCallbackOp.OnStart, memPtr, 0, 0, proc);
}
finally {
leave();
@@ -762,7 +803,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.lifecycleEvent(envPtr, ptr, evt);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.LifecycleOnEvent, ptr, evt, 0, null);
}
finally {
leave();
@@ -779,7 +821,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.memoryReallocate(envPtr, memPtr, cap);
+ PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.MemoryReallocate, memPtr, cap, 0, null);
}
finally {
leave();
@@ -790,13 +833,13 @@ public class PlatformCallbackGateway {
* Initializes native service.
*
* @param memPtr Pointer.
- * @throws org.apache.ignite.IgniteCheckedException In case of error.
+ * @throws IgniteCheckedException In case of error.
*/
public long serviceInit(long memPtr) throws IgniteCheckedException {
enter();
try {
- return PlatformCallbackUtils.serviceInit(envPtr, memPtr);
+ return PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ServiceInit, memPtr);
}
finally {
leave();
@@ -806,15 +849,14 @@ public class PlatformCallbackGateway {
/**
* Executes native service.
*
- * @param svcPtr Pointer to the service in the native platform.
* @param memPtr Stream pointer.
- * @throws org.apache.ignite.IgniteCheckedException In case of error.
+ * @throws IgniteCheckedException In case of error.
*/
- public void serviceExecute(long svcPtr, long memPtr) throws IgniteCheckedException {
+ public void serviceExecute(long memPtr) throws IgniteCheckedException {
enter();
try {
- PlatformCallbackUtils.serviceExecute(envPtr, svcPtr, memPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ServiceExecute, memPtr);
}
finally {
leave();
@@ -824,15 +866,14 @@ public class PlatformCallbackGateway {
/**
* Cancels native service.
*
- * @param svcPtr Pointer to the service in the native platform.
* @param memPtr Stream pointer.
- * @throws org.apache.ignite.IgniteCheckedException In case of error.
+ * @throws IgniteCheckedException In case of error.
*/
- public void serviceCancel(long svcPtr, long memPtr) throws IgniteCheckedException {
+ public void serviceCancel(long memPtr) throws IgniteCheckedException {
enter();
try {
- PlatformCallbackUtils.serviceCancel(envPtr, svcPtr, memPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ServiceCancel, memPtr);
}
finally {
leave();
@@ -842,16 +883,14 @@ public class PlatformCallbackGateway {
/**
* Invokes service method.
*
- * @param svcPtr Pointer to the service in the native platform.
- * @param outMemPtr Output memory pointer.
- * @param inMemPtr Input memory pointer.
- * @throws org.apache.ignite.IgniteCheckedException In case of error.
+ * @param memPtr Memory pointer.
+ * @throws IgniteCheckedException In case of error.
*/
- public void serviceInvokeMethod(long svcPtr, long outMemPtr, long inMemPtr) throws IgniteCheckedException {
+ public void serviceInvokeMethod(long memPtr) throws IgniteCheckedException {
enter();
try {
- PlatformCallbackUtils.serviceInvokeMethod(envPtr, svcPtr, outMemPtr, inMemPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ServiceInvokeMethod, memPtr);
}
finally {
leave();
@@ -867,7 +906,7 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.clusterNodeFilterApply(envPtr, memPtr);
+ return (int)PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.ClusterNodeFilterApply, memPtr);
}
finally {
leave();
@@ -885,7 +924,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.extensionCallbackInLongOutLong(envPtr, typ, arg1);
+ return PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ExtensionInLongOutLong, typ, arg1, 0, null);
}
finally {
leave();
@@ -904,7 +944,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.extensionCallbackInLongLongOutLong(envPtr, typ, arg1, arg2);
+ return PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr,
+ PlatformCallbackOp.ExtensionInLongLongOutLong, typ, arg1, arg2, null);
}
finally {
leave();
@@ -918,7 +959,7 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.onClientDisconnected(envPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.OnClientDisconnected, 0);
}
finally {
leave();
@@ -934,7 +975,8 @@ public class PlatformCallbackGateway {
enter();
try {
- PlatformCallbackUtils.onClientReconnected(envPtr, clusterRestarted);
+ PlatformCallbackUtils.inLongOutLong(envPtr,
+ PlatformCallbackOp.OnClientReconnected, clusterRestarted ? 1 : 0);
}
finally {
leave();
@@ -985,7 +1027,7 @@ public class PlatformCallbackGateway {
public void onStop() {
block();
- PlatformCallbackUtils.onStop(envPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.OnStop, 0);
}
/**
@@ -999,7 +1041,8 @@ public class PlatformCallbackGateway {
enter();
try {
- return PlatformCallbackUtils.affinityFunctionInit(envPtr, memPtr, baseFunc);
+ return PlatformCallbackUtils.inLongLongLongObjectOutLong(envPtr, PlatformCallbackOp.AffinityFunctionInit,
+ memPtr, 0, 0, baseFunc);
}
finally {
leave();
@@ -1009,15 +1052,15 @@ public class PlatformCallbackGateway {
/**
* Gets the partition from affinity function.
*
- * @param ptr Affinity function pointer.
- * @param memPtr Pointer to a stream with key object.
+ * @param memPtr Pointer to a stream with data.
* @return Partition number for a given key.
*/
- public int affinityFunctionPartition(long ptr, long memPtr) {
+ public int affinityFunctionPartition(long memPtr) {
enter();
try {
- return PlatformCallbackUtils.affinityFunctionPartition(envPtr, ptr, memPtr);
+ return (int)PlatformCallbackUtils.inLongOutLong(envPtr,
+ PlatformCallbackOp.AffinityFunctionPartition, memPtr);
}
finally {
leave();
@@ -1027,15 +1070,13 @@ public class PlatformCallbackGateway {
/**
* Assigns the affinity partitions.
*
- * @param ptr Affinity function pointer.
- * @param outMemPtr Pointer to a stream with affinity context.
- * @param inMemPtr Pointer to a stream with result.
+ * @param memPtr Pointer to a stream.
*/
- public void affinityFunctionAssignPartitions(long ptr, long outMemPtr, long inMemPtr){
+ public void affinityFunctionAssignPartitions(long memPtr){
enter();
try {
- PlatformCallbackUtils.affinityFunctionAssignPartitions(envPtr, ptr, outMemPtr, inMemPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.AffinityFunctionAssignPartitions, memPtr);
}
finally {
leave();
@@ -1045,14 +1086,13 @@ public class PlatformCallbackGateway {
/**
* Removes the node from affinity function.
*
- * @param ptr Affinity function pointer.
- * @param memPtr Pointer to a stream with node id.
+ * @param memPtr Pointer to a stream.
*/
- public void affinityFunctionRemoveNode(long ptr, long memPtr) {
+ public void affinityFunctionRemoveNode(long memPtr) {
enter();
try {
- PlatformCallbackUtils.affinityFunctionRemoveNode(envPtr, ptr, memPtr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.AffinityFunctionRemoveNode, memPtr);
}
finally {
leave();
@@ -1069,7 +1109,7 @@ public class PlatformCallbackGateway {
return; // skip: destroy is not necessary during shutdown.
try {
- PlatformCallbackUtils.affinityFunctionDestroy(envPtr, ptr);
+ PlatformCallbackUtils.inLongOutLong(envPtr, PlatformCallbackOp.AffinityFunctionDestroy, ptr);
}
finally {
leave();
@@ -1097,7 +1137,7 @@ public class PlatformCallbackGateway {
/**
* Enter gateway.
*/
- protected boolean tryEnter() {
+ private boolean tryEnter() {
return lock.enterBusy();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackOp.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackOp.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackOp.java
new file mode 100644
index 0000000..973ba51
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackOp.java
@@ -0,0 +1,206 @@
+/*
+ * 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.
+ */
+
+package org.apache.ignite.internal.processors.platform.callback;
+
+/**
+ * Platform callback operation codes.
+ */
+class PlatformCallbackOp {
+ /** */
+ public static final int CacheStoreCreate = 1;
+
+ /** */
+ public static final int CacheStoreInvoke = 2;
+
+ /** */
+ public static final int CacheStoreDestroy = 3;
+
+ /** */
+ public static final int CacheStoreSessionCreate = 4;
+
+ /** */
+ public static final int CacheEntryFilterCreate = 5;
+
+ /** */
+ public static final int CacheEntryFilterApply = 6;
+
+ /** */
+ public static final int CacheEntryFilterDestroy = 7;
+
+ /** */
+ public static final int CacheInvoke = 8;
+
+ /** */
+ public static final int ComputeTaskMap = 9;
+
+ /** */
+ public static final int ComputeTaskJobResult = 10;
+
+ /** */
+ public static final int ComputeTaskReduce = 11;
+
+ /** */
+ public static final int ComputeTaskComplete = 12;
+
+ /** */
+ public static final int ComputeJobSerialize = 13;
+
+ /** */
+ public static final int ComputeJobCreate = 14;
+
+ /** */
+ public static final int ComputeJobExecute = 15;
+
+ /** */
+ public static final int ComputeJobCancel = 16;
+
+ /** */
+ public static final int ComputeJobDestroy = 17;
+
+ /** */
+ public static final int ContinuousQueryListenerApply = 18;
+
+ /** */
+ public static final int ContinuousQueryFilterCreate = 19;
+
+ /** */
+ public static final int ContinuousQueryFilterApply = 20;
+
+ /** */
+ public static final int ContinuousQueryFilterRelease = 21;
+
+ /** */
+ public static final int DataStreamerTopologyUpdate = 22;
+
+ /** */
+ public static final int DataStreamerStreamReceiverInvoke = 23;
+
+ /** */
+ public static final int FutureByteResult = 24;
+
+ /** */
+ public static final int FutureBoolResult = 25;
+
+ /** */
+ public static final int FutureShortResult = 26;
+
+ /** */
+ public static final int FutureCharResult = 27;
+
+ /** */
+ public static final int FutureIntResult = 28;
+
+ /** */
+ public static final int FutureFloatResult = 29;
+
+ /** */
+ public static final int FutureLongResult = 30;
+
+ /** */
+ public static final int FutureDoubleResult = 31;
+
+ /** */
+ public static final int FutureObjectResult = 32;
+
+ /** */
+ public static final int FutureNullResult = 33;
+
+ /** */
+ public static final int FutureError = 34;
+
+ /** */
+ public static final int LifecycleOnEvent = 35;
+
+ /** */
+ public static final int MemoryReallocate = 36;
+
+ /** */
+ public static final int MessagingFilterCreate = 37;
+
+ /** */
+ public static final int MessagingFilterApply = 38;
+
+ /** */
+ public static final int MessagingFilterDestroy = 39;
+
+ /** */
+ public static final int EventFilterCreate = 40;
+
+ /** */
+ public static final int EventFilterApply = 41;
+
+ /** */
+ public static final int EventFilterDestroy = 42;
+
+ /** */
+ public static final int ServiceInit = 43;
+
+ /** */
+ public static final int ServiceExecute = 44;
+
+ /** */
+ public static final int ServiceCancel = 45;
+
+ /** */
+ public static final int ServiceInvokeMethod = 46;
+
+ /** */
+ public static final int ClusterNodeFilterApply = 47;
+
+ /** */
+ public static final int NodeInfo = 48;
+
+ /** */
+ public static final int OnStart = 49;
+
+ /** */
+ public static final int OnStop = 50;
+
+ /** */
+ public static final int ExtensionInLongOutLong = 51;
+
+ /** */
+ public static final int ExtensionInLongLongOutLong = 52;
+
+ /** */
+ public static final int OnClientDisconnected = 53;
+
+ /** */
+ public static final int OnClientReconnected = 54;
+
+ /** */
+ public static final int AffinityFunctionInit = 55;
+
+ /** */
+ public static final int AffinityFunctionPartition = 56;
+
+ /** */
+ public static final int AffinityFunctionAssignPartitions = 57;
+
+ /** */
+ public static final int AffinityFunctionRemoveNode = 58;
+
+ /** */
+ public static final int AffinityFunctionDestroy = 59;
+
+ /** */
+ public static final int ComputeTaskLocalJobResult = 60;
+
+ /** */
+ public static final int ComputeJobExecuteLocal = 61;
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackUtils.java
index 9d60ec0..f823cb3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/callback/PlatformCallbackUtils.java
@@ -17,533 +17,12 @@
package org.apache.ignite.internal.processors.platform.callback;
-import org.apache.ignite.internal.processors.platform.PlatformTargetProxy;
-
/**
* Platform callback utility methods. Implemented in target platform. All methods in this class must be
* package-visible and invoked only through {@link PlatformCallbackGateway}.
*/
public class PlatformCallbackUtils {
/**
- * Create cache store.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer.
- */
- static native long cacheStoreCreate(long envPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Object pointer.
- * @param memPtr Memory pointer.
- * @return Result.
- */
- static native int cacheStoreInvoke(long envPtr, long objPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Object pointer.
- */
- static native void cacheStoreDestroy(long envPtr, long objPtr);
-
- /**
- * Creates cache store session.
- *
- * @param envPtr Environment pointer.
- * @param storePtr Store instance pointer.
- * @return Session instance pointer.
- */
- static native long cacheStoreSessionCreate(long envPtr, long storePtr);
-
- /**
- * Creates cache entry filter and returns a pointer.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer.
- */
- static native long cacheEntryFilterCreate(long envPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- * @param memPtr Memory pointer.
- * @return Result.
- */
- static native int cacheEntryFilterApply(long envPtr, long objPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- */
- static native void cacheEntryFilterDestroy(long envPtr, long objPtr);
-
- /**
- * Invoke cache entry processor.
- *
- * @param envPtr Environment pointer.
- * @param outMemPtr Output memory pointer.
- * @param inMemPtr Input memory pointer.
- */
- static native void cacheInvoke(long envPtr, long outMemPtr, long inMemPtr);
-
- /**
- * Perform native task map. Do not throw exceptions, serializing them to the output stream instead.
- *
- * @param envPtr Environment pointer.
- * @param taskPtr Task pointer.
- * @param outMemPtr Output memory pointer (exists if topology changed, otherwise {@code 0}).
- * @param inMemPtr Input memory pointer.
- */
- static native void computeTaskMap(long envPtr, long taskPtr, long outMemPtr, long inMemPtr);
-
- /**
- * Perform native task job result notification.
- *
- * @param envPtr Environment pointer.
- * @param taskPtr Task pointer.
- * @param jobPtr Job pointer.
- * @param memPtr Memory pointer (always zero for local job execution).
- * @return Job result enum ordinal.
- */
- static native int computeTaskJobResult(long envPtr, long taskPtr, long jobPtr, long memPtr);
-
- /**
- * Perform native task reduce.
- *
- * @param envPtr Environment pointer.
- * @param taskPtr Task pointer.
- */
- static native void computeTaskReduce(long envPtr, long taskPtr);
-
- /**
- * Complete task with native error.
- *
- * @param envPtr Environment pointer.
- * @param taskPtr Task pointer.
- * @param memPtr Memory pointer with exception data or {@code 0} in case of success.
- */
- static native void computeTaskComplete(long envPtr, long taskPtr, long memPtr);
-
- /**
- * Serialize native job.
- *
- * @param envPtr Environment pointer.
- * @param jobPtr Job pointer.
- * @param memPtr Memory pointer.
- * @return {@code True} if serialization succeeded.
- */
- static native int computeJobSerialize(long envPtr, long jobPtr, long memPtr);
-
- /**
- * Create job in native platform.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer to job.
- */
- static native long computeJobCreate(long envPtr, long memPtr);
-
- /**
- * Execute native job on a node other than where it was created.
- *
- * @param envPtr Environment pointer.
- * @param jobPtr Job pointer.
- * @param cancel Cancel flag.
- * @param memPtr Memory pointer to write result to for remote job execution or {@code 0} for local job execution.
- */
- static native void computeJobExecute(long envPtr, long jobPtr, int cancel, long memPtr);
-
- /**
- * Cancel the job.
- *
- * @param envPtr Environment pointer.
- * @param jobPtr Job pointer.
- */
- static native void computeJobCancel(long envPtr, long jobPtr);
-
- /**
- * Destroy the job.
- *
- * @param envPtr Environment pointer.
- * @param ptr Pointer.
- */
- static native void computeJobDestroy(long envPtr, long ptr);
-
- /**
- * Invoke local callback.
- *
- * @param envPtr Environment pointer.
- * @param cbPtr Callback pointer.
- * @param memPtr Memory pointer.
- */
- static native void continuousQueryListenerApply(long envPtr, long cbPtr, long memPtr);
-
- /**
- * Create filter in native platform.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer to created filter.
- */
- static native long continuousQueryFilterCreate(long envPtr, long memPtr);
-
- /**
- * Invoke remote filter.
- *
- * @param envPtr Environment pointer.
- * @param filterPtr Filter pointer.
- * @param memPtr Memory pointer.
- * @return Result.
- */
- static native int continuousQueryFilterApply(long envPtr, long filterPtr, long memPtr);
-
- /**
- * Release remote filter.
- *
- * @param envPtr Environment pointer.
- * @param filterPtr Filter pointer.
- */
- static native void continuousQueryFilterRelease(long envPtr, long filterPtr);
-
- /**
- * Notify native data streamer about topology update.
- *
- * @param envPtr Environment pointer.
- * @param ptr Data streamer native pointer.
- * @param topVer Topology version.
- * @param topSize Topology size.
- */
- static native void dataStreamerTopologyUpdate(long envPtr, long ptr, long topVer, int topSize);
-
- /**
- * Invoke stream receiver.
- *
- * @param envPtr Environment pointer.
- * @param ptr Receiver native pointer.
- * @param cache Cache object.
- * @param memPtr Stream pointer.
- * @param keepBinary Binary flag.
- */
- static native void dataStreamerStreamReceiverInvoke(long envPtr, long ptr, PlatformTargetProxy cache, long memPtr,
- boolean keepBinary);
-
- /**
- * Notify future with byte result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureByteResult(long envPtr, long futPtr, int res);
-
- /**
- * Notify future with boolean result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureBoolResult(long envPtr, long futPtr, int res);
-
- /**
- * Notify future with short result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureShortResult(long envPtr, long futPtr, int res);
-
- /**
- * Notify future with byte result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureCharResult(long envPtr, long futPtr, int res);
-
- /**
- * Notify future with int result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureIntResult(long envPtr, long futPtr, int res);
-
- /**
- * Notify future with float result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureFloatResult(long envPtr, long futPtr, float res);
-
- /**
- * Notify future with long result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureLongResult(long envPtr, long futPtr, long res);
-
- /**
- * Notify future with double result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param res Result.
- */
- static native void futureDoubleResult(long envPtr, long futPtr, double res);
-
- /**
- * Notify future with object result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param memPtr Memory pointer.
- */
- static native void futureObjectResult(long envPtr, long futPtr, long memPtr);
-
- /**
- * Notify future with null result.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- */
- static native void futureNullResult(long envPtr, long futPtr);
-
- /**
- * Notify future with error.
- *
- * @param envPtr Environment pointer.
- * @param futPtr Future pointer.
- * @param memPtr Pointer to memory with error information.
- */
- static native void futureError(long envPtr, long futPtr, long memPtr);
-
- /**
- * Creates message filter and returns a pointer.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer.
- */
- static native long messagingFilterCreate(long envPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- * @param memPtr Memory pointer.
- * @return Result.
- */
- static native int messagingFilterApply(long envPtr, long objPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- */
- static native void messagingFilterDestroy(long envPtr, long objPtr);
-
- /**
- * Creates event filter and returns a pointer.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Memory pointer.
- * @return Pointer.
- */
- static native long eventFilterCreate(long envPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- * @param memPtr Memory pointer.
- * @return Result.
- */
- static native int eventFilterApply(long envPtr, long objPtr, long memPtr);
-
- /**
- * @param envPtr Environment pointer.
- * @param objPtr Pointer.
- */
- static native void eventFilterDestroy(long envPtr, long objPtr);
-
- /**
- * Sends node info to native target.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Ptr to a stream with serialized node.
- */
- static native void nodeInfo(long envPtr, long memPtr);
-
- /**
- * Kernal start callback.
- *
- * @param envPtr Environment pointer.
- * @param proc Platform processor.
- * @param memPtr Memory pointer.
- */
- static native void onStart(long envPtr, Object proc, long memPtr);
-
- /*
- * Kernal stop callback.
- *
- * @param envPtr Environment pointer.
- */
- static native void onStop(long envPtr);
-
- /**
- * Lifecycle event callback.
- *
- * @param envPtr Environment pointer.
- * @param ptr Holder pointer.
- * @param evt Event.
- */
- static native void lifecycleEvent(long envPtr, long ptr, int evt);
-
- /**
- * Re-allocate external memory chunk.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Cross-platform pointer.
- * @param cap Capacity.
- */
- static native void memoryReallocate(long envPtr, long memPtr, int cap);
-
- /**
- * Initializes native service.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Stream pointer.
- * @return Pointer to the native platform service.
- */
- static native long serviceInit(long envPtr, long memPtr);
-
- /**
- * Executes native service.
- *
- * @param envPtr Environment pointer.
- * @param svcPtr Pointer to the service in the native platform.
- * @param memPtr Stream pointer.
- */
- static native void serviceExecute(long envPtr, long svcPtr, long memPtr);
-
- /**
- * Cancels native service.
- *
- * @param envPtr Environment pointer.
- * @param svcPtr Pointer to the service in the native platform.
- * @param memPtr Stream pointer.
- */
- static native void serviceCancel(long envPtr, long svcPtr, long memPtr);
-
- /**
- * Invokes service method.
- *
- * @param envPtr Environment pointer.
- * @param svcPtr Pointer to the service in the native platform.
- * @param outMemPtr Output memory pointer.
- * @param inMemPtr Input memory pointer.
- */
- static native void serviceInvokeMethod(long envPtr, long svcPtr, long outMemPtr, long inMemPtr);
-
- /**
- * Invokes cluster node filter.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Stream pointer.
- */
- static native int clusterNodeFilterApply(long envPtr, long memPtr);
-
- /**
- * Extension callback accepting single long argument and returning long result.
- *
- * @param envPtr Environment pointer.
- * @param typ Operation type.
- * @param arg1 Argument 1.
- * @return Long result.
- */
- static native long extensionCallbackInLongOutLong(long envPtr, int typ, long arg1);
-
- /**
- * Extension callback accepting two long arguments and returning long result.
- *
- * @param envPtr Environment pointer.
- * @param typ Operation type.
- * @param arg1 Argument 1.
- * @param arg2 Argument 2.
- * @return Long result.
- */
- static native long extensionCallbackInLongLongOutLong(long envPtr, int typ, long arg1, long arg2);
-
- /**
- * Notifies platform about client disconnect.
- *
- * @param envPtr Environment pointer.
- */
- static native void onClientDisconnected(long envPtr);
-
- /**
- * Notifies platform about client reconnect.
- *
- * @param envPtr Environment pointer.
- * @param clusterRestarted Cluster restarted flag.
- */
- static native void onClientReconnected(long envPtr, boolean clusterRestarted);
-
- /**
- * Initializes affinity function.
- *
- * @param envPtr Environment pointer.
- * @param memPtr Pointer to a stream with serialized affinity function.
- * @param baseFunc Optional func for base calls.
- * @return Affinity function pointer.
- */
- static native long affinityFunctionInit(long envPtr, long memPtr, PlatformTargetProxy baseFunc);
-
- /**
- * Gets the partition from affinity function.
- *
- * @param envPtr Environment pointer.
- * @param ptr Affinity function pointer.
- * @param memPtr Pointer to a stream with key object.
- * @return Partition number for a given key.
- */
- static native int affinityFunctionPartition(long envPtr, long ptr, long memPtr);
-
- /**
- * Assigns the affinity partitions.
- *
- * @param envPtr Environment pointer.
- * @param ptr Affinity function pointer.
- * @param outMemPtr Pointer to a stream with affinity context.
- * @param inMemPtr Pointer to a stream with result.
- */
- static native void affinityFunctionAssignPartitions(long envPtr, long ptr, long outMemPtr, long inMemPtr);
-
- /**
- * Removes the node from affinity function.
- *
- * @param envPtr Environment pointer.
- * @param ptr Affinity function pointer.
- * @param memPtr Pointer to a stream with node id.
- */
- static native void affinityFunctionRemoveNode(long envPtr, long ptr, long memPtr);
-
- /**
- * Destroys the affinity function.
- *
- * @param envPtr Environment pointer.
- * @param ptr Affinity function pointer.
- */
- static native void affinityFunctionDestroy(long envPtr, long ptr);
-
- /**
* Redirects the console output.
*
* @param str String to write.
@@ -572,6 +51,29 @@ public class PlatformCallbackUtils {
static native boolean loggerIsLevelEnabled(long envPtr, int level);
/**
+ * Performs a generic long-long operation.
+ *
+ * @param envPtr Environment pointer.
+ * @param type Operation code.
+ * @param val Value.
+ * @return Value.
+ */
+ static native long inLongOutLong(long envPtr, int type, long val);
+
+ /**
+ * Performs a generic out-in operation.
+ *
+ * @param envPtr Environment pointer.
+ * @param type Operation code.
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @param val3 Third value.
+ * @param arg Object argument.
+ * @return Value.
+ */
+ static native long inLongLongLongObjectOutLong(long envPtr, int type, long val1, long val2, long val3, Object arg);
+
+ /**
* Private constructor.
*/
private PlatformCallbackUtils() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractJob.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractJob.java
index 32aed39..56875c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractJob.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractJob.java
@@ -131,7 +131,7 @@ public abstract class PlatformAbstractJob implements PlatformJob, Externalizable
// Local job, must execute it with respect to possible concurrent task completion.
if (task.onJobLock()) {
try {
- ctx.gateway().computeJobExecute(ptr, cancel ? 1 : 0, 0);
+ ctx.gateway().computeJobExecuteLocal(ptr, cancel ? 1 : 0);
return LOC_JOB_RES;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractTask.java
index fe1e316..6a9fed5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformAbstractTask.java
@@ -65,8 +65,6 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
@Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) {
assert rcvd.isEmpty() : "Should not cache result in Java for interop task";
- int plc;
-
lock.readLock().lock();
try {
@@ -78,9 +76,11 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
Object res0bj = res.getData();
+ int plc;
+
if (res0bj == PlatformAbstractJob.LOC_JOB_RES)
// Processing local job execution result.
- plc = ctx.gateway().computeTaskJobResult(taskPtr, job.pointer(), 0);
+ plc = ctx.gateway().computeTaskLocalJobResult(taskPtr, job.pointer());
else {
// Processing remote job execution result or exception.
try (PlatformMemory mem = ctx.memory().allocate()) {
@@ -88,6 +88,9 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(taskPtr);
+ writer.writeLong(job.pointer());
+
writer.writeUuid(res.getNode().id());
writer.writeBoolean(res.isCancelled());
@@ -97,7 +100,7 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
out.synchronize();
- plc = ctx.gateway().computeTaskJobResult(taskPtr, job.pointer(), mem.pointer());
+ plc = ctx.gateway().computeTaskJobResult(mem.pointer());
}
}
@@ -184,7 +187,7 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
* @return {@code} True if task is not completed yet, {@code false} otherwise.
*/
@SuppressWarnings("LockAcquiredButNotSafelyReleased")
- public boolean onJobLock() {
+ boolean onJobLock() {
lock.readLock().lock();
if (done) {
@@ -199,7 +202,7 @@ public abstract class PlatformAbstractTask implements ComputeTask<Object, Void>
/**
* Callback invoked by job when task can be unlocked.
*/
- public void onJobUnlock() {
+ void onJobUnlock() {
assert !done;
lock.readLock().unlock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformClosureJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformClosureJob.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformClosureJob.java
index f8567ce..25926eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformClosureJob.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformClosureJob.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.processors.platform.PlatformContext;
import org.apache.ignite.internal.processors.platform.memory.PlatformInputStream;
import org.apache.ignite.internal.processors.platform.memory.PlatformMemory;
+import org.apache.ignite.internal.processors.platform.memory.PlatformOutputStream;
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
import org.jetbrains.annotations.Nullable;
@@ -63,9 +64,16 @@ public class PlatformClosureJob extends PlatformAbstractJob {
createJob(ctx);
try (PlatformMemory mem = ctx.memory().allocate()) {
- PlatformInputStream in = mem.input();
+ PlatformOutputStream out = mem.output();
+
+ out.writeLong(ptr);
+ out.writeBoolean(false); // cancel
+
+ out.synchronize();
- ctx.gateway().computeJobExecute(ptr, 0, mem.pointer());
+ ctx.gateway().computeJobExecute(mem.pointer());
+
+ PlatformInputStream in = mem.input();
in.synchronize();
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullJob.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullJob.java
index 51c9cdb..9ff9609 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullJob.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullJob.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.platform.compute;
+import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
@@ -26,6 +27,7 @@ import org.apache.ignite.internal.processors.platform.PlatformContext;
import org.apache.ignite.internal.processors.platform.PlatformProcessor;
import org.apache.ignite.internal.processors.platform.memory.PlatformInputStream;
import org.apache.ignite.internal.processors.platform.memory.PlatformMemory;
+import org.apache.ignite.internal.processors.platform.memory.PlatformOutputStream;
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
import org.jetbrains.annotations.Nullable;
@@ -64,7 +66,7 @@ public class PlatformFullJob extends PlatformAbstractJob {
private transient byte state;
/**
- * {@link java.io.Externalizable} support.
+ * {@link Externalizable} support.
*/
@SuppressWarnings("UnusedDeclaration")
public PlatformFullJob() {
@@ -114,9 +116,16 @@ public class PlatformFullJob extends PlatformAbstractJob {
return runLocal(ctx, cancel);
else {
try (PlatformMemory mem = ctx.memory().allocate()) {
- PlatformInputStream in = mem.input();
+ PlatformOutputStream out = mem.output();
+
+ out.writeLong(ptr);
+ out.writeBoolean(cancel); // cancel
+
+ out.synchronize();
- ctx.gateway().computeJobExecute(ptr, cancel ? 1 : 0, mem.pointer());
+ ctx.gateway().computeJobExecute(mem.pointer());
+
+ PlatformInputStream in = mem.input();
in.synchronize();
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullTask.java
index e2f6720..3134066 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformFullTask.java
@@ -80,26 +80,26 @@ public final class PlatformFullTask extends PlatformAbstractTask {
PlatformMemoryManager memMgr = ctx.memory();
- try (PlatformMemory outMem = memMgr.allocate()) {
- PlatformOutputStream out = outMem.output();
+ try (PlatformMemory mem = memMgr.allocate()) {
+ PlatformOutputStream out = mem.output();
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(taskPtr);
+
write(writer, nodes, subgrid);
out.synchronize();
- try (PlatformMemory inMem = memMgr.allocate()) {
- PlatformInputStream in = inMem.input();
+ ctx.gateway().computeTaskMap(mem.pointer());
- ctx.gateway().computeTaskMap(taskPtr, outMem.pointer(), inMem.pointer());
+ PlatformInputStream in = mem.input();
- in.synchronize();
+ in.synchronize();
- BinaryRawReaderEx reader = ctx.reader(in);
+ BinaryRawReaderEx reader = ctx.reader(in);
- return read(reader, nodes);
- }
+ return read(reader, nodes);
}
}
finally {
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/datastreamer/PlatformStreamReceiverImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/datastreamer/PlatformStreamReceiverImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/datastreamer/PlatformStreamReceiverImpl.java
index d0992fc..c3dde26 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/datastreamer/PlatformStreamReceiverImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/datastreamer/PlatformStreamReceiverImpl.java
@@ -78,6 +78,9 @@ public class PlatformStreamReceiverImpl extends PlatformAbstractPredicate implem
try (PlatformMemory mem = ctx.memory().allocate()) {
PlatformOutputStream out = mem.output();
+ out.writeLong(ptr);
+ out.writeBoolean(keepBinary);
+
BinaryRawWriterEx writer = ctx.writer(out);
writer.writeObject(pred);
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
index 3563dd6..5257b26 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
@@ -396,7 +396,7 @@ public class PlatformDotNetCacheStore<K, V> implements CacheStore<K, V>, Platfor
if (sesPtr == null) {
// Session is not deployed yet, do that.
- sesPtr = platformCtx.gateway().cacheStoreSessionCreate(ptr);
+ sesPtr = platformCtx.gateway().cacheStoreSessionCreate();
ses.properties().put(KEY_SES, sesPtr);
}
@@ -419,11 +419,13 @@ public class PlatformDotNetCacheStore<K, V> implements CacheStore<K, V>, Platfor
BinaryRawWriterEx writer = platformCtx.writer(out);
+ writer.writeLong(ptr);
+
task.apply(writer);
out.synchronize();
- int res = platformCtx.gateway().cacheStoreInvoke(ptr, mem.pointer());
+ int res = platformCtx.gateway().cacheStoreInvoke(mem.pointer());
if (res != 0) {
// Read error
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformAbstractService.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformAbstractService.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformAbstractService.java
index d6a6e16..4db01cf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformAbstractService.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformAbstractService.java
@@ -110,13 +110,15 @@ public abstract class PlatformAbstractService implements PlatformService, Extern
BinaryRawWriterEx writer = platformCtx.writer(out);
+ writer.writeLong(ptr);
+
writer.writeBoolean(srvKeepBinary);
writeServiceContext(ctx, writer);
out.synchronize();
- platformCtx.gateway().serviceExecute(ptr, mem.pointer());
+ platformCtx.gateway().serviceExecute(mem.pointer());
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
@@ -133,13 +135,15 @@ public abstract class PlatformAbstractService implements PlatformService, Extern
BinaryRawWriterEx writer = platformCtx.writer(out);
+ writer.writeLong(ptr);
+
writer.writeBoolean(srvKeepBinary);
writeServiceContext(ctx, writer);
out.synchronize();
- platformCtx.gateway().serviceCancel(ptr, mem.pointer());
+ platformCtx.gateway().serviceCancel(mem.pointer());
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
@@ -173,10 +177,11 @@ public abstract class PlatformAbstractService implements PlatformService, Extern
assert ptr != 0;
assert platformCtx != null;
- try (PlatformMemory outMem = platformCtx.memory().allocate()) {
- PlatformOutputStream out = outMem.output();
+ try (PlatformMemory mem = platformCtx.memory().allocate()) {
+ PlatformOutputStream out = mem.output();
BinaryRawWriterEx writer = platformCtx.writer(out);
+ writer.writeLong(ptr);
writer.writeBoolean(srvKeepBinary);
writer.writeString(mthdName);
@@ -192,17 +197,15 @@ public abstract class PlatformAbstractService implements PlatformService, Extern
out.synchronize();
- try (PlatformMemory inMem = platformCtx.memory().allocate()) {
- PlatformInputStream in = inMem.input();
+ platformCtx.gateway().serviceInvokeMethod(mem.pointer());
- platformCtx.gateway().serviceInvokeMethod(ptr, outMem.pointer(), inMem.pointer());
+ PlatformInputStream in = mem.input();
- in.synchronize();
+ in.synchronize();
- BinaryRawReaderEx reader = platformCtx.reader(in);
+ BinaryRawReaderEx reader = platformCtx.reader(in);
- return PlatformUtils.readInvocationResult(platformCtx, reader);
- }
+ return PlatformUtils.readInvocationResult(platformCtx, reader);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
index e81f4c6..b84744c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
@@ -250,7 +250,7 @@ public class PlatformFutureUtils {
break;
case TYP_FLOAT:
- gate.futureFloatResult(futPtr, (float) res);
+ gate.futureFloatResult(futPtr, Float.floatToIntBits((float) res));
break;
@@ -260,7 +260,7 @@ public class PlatformFutureUtils {
break;
case TYP_DOUBLE:
- gate.futureDoubleResult(futPtr, (double) res);
+ gate.futureDoubleResult(futPtr, Double.doubleToLongBits((double)res));
break;
http://git-wip-us.apache.org/repos/asf/ignite/blob/72ac53da/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformUtils.java
index 0d30ad9..4c0eab4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformUtils.java
@@ -501,6 +501,8 @@ public class PlatformUtils {
BinaryRawWriterEx writer = ctx.writer(out);
+ writer.writeLong(lsnrPtr);
+
int cntPos = writer.reserveInt();
int cnt = 0;
@@ -515,7 +517,7 @@ public class PlatformUtils {
out.synchronize();
- ctx.gateway().continuousQueryListenerApply(lsnrPtr, mem.pointer());
+ ctx.gateway().continuousQueryListenerApply(mem.pointer());
}
catch (Exception e) {
throw toCacheEntryListenerException(e);
@@ -538,11 +540,13 @@ public class PlatformUtils {
try (PlatformMemory mem = ctx.memory().allocate()) {
PlatformOutputStream out = mem.output();
+ out.writeLong(filterPtr);
+
writeCacheEntryEvent(ctx.writer(out), evt);
out.synchronize();
- return ctx.gateway().continuousQueryFilterApply(filterPtr, mem.pointer()) == 1;
+ return ctx.gateway().continuousQueryFilterApply(mem.pointer()) == 1;
}
catch (Exception e) {
throw toCacheEntryListenerException(e);