You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by we...@apache.org on 2015/07/28 04:35:58 UTC
incubator-reef git commit: [REEF-478] Introduce caching in NameClient
LookUp function
Repository: incubator-reef
Updated Branches:
refs/heads/master cd91f6312 -> e5aef1616
[REEF-478] Introduce caching in NameClient LookUp function
This addressed the issue by
* introducing NameCache that caches IPEndPoint and internally used .Net Memory
Cache
* introducing another function in NameClient CacheLookUp() that looks from
cache.
JIRA:
[REEF-478](https://issues.apache.org/jira/browse/REEF-478)
Pull Request:
This closes #318
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/e5aef161
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/e5aef161
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/e5aef161
Branch: refs/heads/master
Commit: e5aef16163553e11f28ce80fbe61062a66154f1f
Parents: cd91f63
Author: Dhruv <dh...@gmail.com>
Authored: Thu Jul 16 00:06:24 2015 -0700
Committer: Markus Weimer <we...@apache.org>
Committed: Mon Jul 27 16:49:49 2015 -0700
----------------------------------------------------------------------
.../cs/Org.Apache.REEF.Common/Io/INameClient.cs | 9 ++
.../NamingService/NameServerTests.cs | 56 ++++++++--
.../Org.Apache.REEF.Network/Naming/NameCache.cs | 105 +++++++++++++++++++
.../Naming/NameClient.cs | 59 ++++++++++-
.../Naming/Parameters/NameCacheConfiguration.cs | 41 ++++++++
.../NetworkService/NsConnection.cs | 2 +-
.../Org.Apache.REEF.Network.csproj | 3 +
7 files changed, 264 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Common/Io/INameClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Common/Io/INameClient.cs b/lang/cs/Org.Apache.REEF.Common/Io/INameClient.cs
index 8fc1f2d..2fb8b45 100644
--- a/lang/cs/Org.Apache.REEF.Common/Io/INameClient.cs
+++ b/lang/cs/Org.Apache.REEF.Common/Io/INameClient.cs
@@ -53,6 +53,15 @@ namespace Org.Apache.REEF.Common.Io
IPEndPoint Lookup(string id);
/// <summary>
+ /// Looks up the IPEndpoint for the registered identifier.
+ /// Use cache if it has entry
+ /// </summary>
+ /// <param name="id">The identifier to look up</param>
+ /// <returns>The mapped IPEndpoint for the identifier, or null if
+ /// the identifier has not been registered with the NameService</returns>
+ IPEndPoint CacheLookup(string id);
+
+ /// <summary>
/// Looks up the IPEndpoint for each of the registered identifiers in the list.
/// </summary>
/// <param name="ids">The list of identifiers to look up</param>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network.Tests/NamingService/NameServerTests.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Tests/NamingService/NameServerTests.cs b/lang/cs/Org.Apache.REEF.Network.Tests/NamingService/NameServerTests.cs
index 6523e84..b11023a 100644
--- a/lang/cs/Org.Apache.REEF.Network.Tests/NamingService/NameServerTests.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Tests/NamingService/NameServerTests.cs
@@ -19,11 +19,14 @@
using System;
using System.Collections.Generic;
+using System.Globalization;
using System.Net;
using System.Threading;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Examples.Tasks.StreamingTasks;
using Org.Apache.REEF.Network.Naming;
+using Org.Apache.REEF.Network.Naming.Parameters;
using Org.Apache.REEF.Tang.Annotations;
using Org.Apache.REEF.Tang.Implementations.Tang;
using Org.Apache.REEF.Tang.Interface;
@@ -48,8 +51,8 @@ namespace Org.Apache.REEF.Network.Tests.NamingService
{
using (var server = BuildNameServer())
{
- var nameClient = new NameClient(server.LocalEndpoint);
- var nameClient2 = new NameClient(server.LocalEndpoint);
+ var nameClient = GetNameClientInstance(server.LocalEndpoint);
+ var nameClient2 = GetNameClientInstance(server.LocalEndpoint);
nameClient2.Register("1", new IPEndPoint(IPAddress.Any, 8080));
nameClient.Lookup("1");
}
@@ -60,8 +63,8 @@ namespace Org.Apache.REEF.Network.Tests.NamingService
{
using (var server = BuildNameServer())
{
- var nameClient = new NameClient(server.LocalEndpoint);
- var nameClient2 = new NameClient(server.LocalEndpoint);
+ var nameClient = GetNameClientInstance(server.LocalEndpoint);
+ var nameClient2 = GetNameClientInstance(server.LocalEndpoint);
nameClient2.Register("1", new IPEndPoint(IPAddress.Any, 8080));
nameClient.Lookup("1");
}
@@ -72,8 +75,8 @@ namespace Org.Apache.REEF.Network.Tests.NamingService
{
using (var server = BuildNameServer())
{
- var nameClient = new NameClient(server.LocalEndpoint);
- var nameClient2 = new NameClient(server.LocalEndpoint);
+ var nameClient = GetNameClientInstance(server.LocalEndpoint);
+ var nameClient2 = GetNameClientInstance(server.LocalEndpoint);
nameClient.Register("1", new IPEndPoint(IPAddress.Any, 8080));
nameClient2.Lookup("1");
}
@@ -233,6 +236,33 @@ namespace Org.Apache.REEF.Network.Tests.NamingService
}
}
+ [TestMethod]
+ public void TestNameCache()
+ {
+ double interval = 50;
+ var config =
+ TangFactory.GetTang()
+ .NewConfigurationBuilder()
+ .BindNamedParameter<NameCacheConfiguration.CacheEntryExpiryTime, double>(
+ GenericType<NameCacheConfiguration.CacheEntryExpiryTime>.Class,
+ interval.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ var injector = TangFactory.GetTang().NewInjector(config);
+ var cache = injector.GetInstance<NameCache>();
+
+ cache.Set("dst1", new IPEndPoint(IPAddress.Any, 0));
+ Thread.Sleep(100);
+ var value = cache.Get("dst1");
+ Assert.IsNull(value);
+
+ IPAddress address = new IPAddress(1234);
+ cache.Set("dst1", new IPEndPoint(address, 0));
+ value = cache.Get("dst1");
+ Assert.IsNotNull(value);
+ Assert.AreEqual(address, value.Address);
+ }
+
public static INameServer BuildNameServer(int listenPort = 0)
{
var builder = TangFactory.GetTang()
@@ -254,6 +284,20 @@ namespace Org.Apache.REEF.Network.Tests.NamingService
return TangFactory.GetTang().NewInjector(nameClientConfiguration).GetInstance<NameClient>();
}
+ private NameClient GetNameClientInstance(IPEndPoint endPoint)
+ {
+ var config1 = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindNamedParameter<NamingConfigurationOptions.NameServerAddress, string>(
+ GenericType<NamingConfigurationOptions.NameServerAddress>.Class,
+ endPoint.Address.ToString())
+ .BindNamedParameter<NamingConfigurationOptions.NameServerPort, int>(
+ GenericType<NamingConfigurationOptions.NameServerPort>.Class,
+ endPoint.Port.ToString())
+ .Build();
+
+ return TangFactory.GetTang().NewInjector(config1).GetInstance<NameClient>();
+ }
+
private class ConstructorInjection
{
[Inject]
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network/Naming/NameCache.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Naming/NameCache.cs b/lang/cs/Org.Apache.REEF.Network/Naming/NameCache.cs
new file mode 100644
index 0000000..9757915
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Naming/NameCache.cs
@@ -0,0 +1,105 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Specialized;
+using System.Net;
+using System.Runtime.Caching;
+using Org.Apache.REEF.Network.Naming.Parameters;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Naming
+{
+ /// <summary>
+ /// Cache class for caching IpEndPoint Lookups
+ /// </summary>
+ internal class NameCache
+ {
+ private readonly MemoryCache _cache;
+
+ /// <summary>
+ /// Duration in milli seconds after which cache entry expires
+ /// Usage in cache requires it to be double than int or long
+ /// </summary>
+ private readonly double _expirationDuration;
+
+ [Inject]
+ private NameCache(
+ [Parameter(typeof (NameCacheConfiguration.CacheEntryExpiryTime))] double expirationDuration,
+ [Parameter(typeof (NameCacheConfiguration.CacheMemoryLimit))] string memoryLimit,
+ [Parameter(typeof (NameCacheConfiguration.PollingInterval))] string pollingInterval)
+ {
+ var config = new NameValueCollection
+ {
+ {"pollingInterval", pollingInterval},
+ {"physicalMemoryLimitPercentage", "0"},
+ {"cacheMemoryLimitMegabytes", memoryLimit}
+ };
+
+ _cache = new MemoryCache("NameClientCache", config);
+ _expirationDuration = expirationDuration;
+ }
+
+ /// <summary>
+ /// Add an entry to cache if it does not exist or replace if it already ecists
+ /// </summary>
+ /// <param name="identifier">remote destination Id</param>
+ /// <param name="value">IPEndPoint of remote destination</param>
+ internal void Set(string identifier, IPEndPoint value)
+ {
+ _cache.Set(identifier, value, DateTimeOffset.Now.AddMilliseconds(_expirationDuration));
+ }
+
+ /// <summary>
+ /// Gets the cached remote IpEndPoint given the name
+ /// </summary>
+ /// <param name="identifier">Remote destination name/Id</param>
+ /// <returns>IPEndPoint of remote destination if it is cached, null otherwise</returns>
+ internal IPEndPoint Get(string identifier)
+ {
+ var entry = _cache.Get(identifier);
+ return entry as IPEndPoint;
+ }
+
+ /// <summary>
+ /// Removes the entry from the cache
+ /// </summary>
+ /// <param name="identifier"></param>
+ internal void RemoveEntry(string identifier)
+ {
+ _cache.Remove(identifier);
+ }
+
+ /// <summary>
+ /// returns physical memory of the cache in MB
+ /// </summary>
+ internal long PhysicalMemoryLimit
+ {
+ get { return _cache.CacheMemoryLimit; }
+ }
+
+ /// <summary>
+ /// returns the interval after which Cache checks its memory usage
+ /// </summary>
+ internal TimeSpan PollingInterval
+ {
+ get { return _cache.PollingInterval; }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network/Naming/NameClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Naming/NameClient.cs b/lang/cs/Org.Apache.REEF.Network/Naming/NameClient.cs
index a858879..4629ad1 100644
--- a/lang/cs/Org.Apache.REEF.Network/Naming/NameClient.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Naming/NameClient.cs
@@ -27,6 +27,7 @@ using Org.Apache.REEF.Common.Io;
using Org.Apache.REEF.Network.Naming.Codec;
using Org.Apache.REEF.Network.Naming.Events;
using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Implementations.Tang;
using Org.Apache.REEF.Utilities.Diagnostics;
using Org.Apache.REEF.Utilities.Logging;
using Org.Apache.REEF.Wake;
@@ -54,8 +55,8 @@ namespace Org.Apache.REEF.Network.Naming
private NameLookupClient _lookupClient;
private NameRegisterClient _registerClient;
-
private bool _disposed;
+ private readonly NameCache _cache;
/// <summary>
/// Constructs a NameClient to register, lookup, and unregister IPEndpoints
@@ -63,14 +64,35 @@ namespace Org.Apache.REEF.Network.Naming
/// </summary>
/// <param name="remoteAddress">The ip address of the NameServer</param>
/// <param name="remotePort">The port of the NameServer</param>
+ [Obsolete("This constructor will be made private in 0.13 version", false)]
[Inject]
public NameClient(
- [Parameter(typeof(NamingConfigurationOptions.NameServerAddress))] string remoteAddress,
- [Parameter(typeof(NamingConfigurationOptions.NameServerPort))] int remotePort)
+ [Parameter(typeof (NamingConfigurationOptions.NameServerAddress))] string remoteAddress,
+ [Parameter(typeof (NamingConfigurationOptions.NameServerPort))] int remotePort)
+ {
+ IPEndPoint remoteEndpoint = new IPEndPoint(IPAddress.Parse(remoteAddress), remotePort);
+ Initialize(remoteEndpoint);
+ _disposed = false;
+ _cache = TangFactory.GetTang().NewInjector().GetInstance<NameCache>();
+ }
+
+ /// <summary>
+ /// Constructs a NameClient to register, lookup, and unregister IPEndpoints
+ /// with the NameServer.
+ /// </summary>
+ /// <param name="remoteAddress">The ip address of the NameServer</param>
+ /// <param name="remotePort">The port of the NameServer</param>
+ /// <param name="cache">The NameCache for caching IpAddresses</param>
+ [Inject]
+ private NameClient(
+ [Parameter(typeof(NamingConfigurationOptions.NameServerAddress))] string remoteAddress,
+ [Parameter(typeof(NamingConfigurationOptions.NameServerPort))] int remotePort,
+ NameCache cache)
{
IPEndPoint remoteEndpoint = new IPEndPoint(IPAddress.Parse(remoteAddress), remotePort);
Initialize(remoteEndpoint);
_disposed = false;
+ _cache = cache;
}
/// <summary>
@@ -78,9 +100,11 @@ namespace Org.Apache.REEF.Network.Naming
/// with the NameServer.
/// </summary>
/// <param name="remoteEndpoint">The endpoint of the NameServer</param>
- public NameClient(IPEndPoint remoteEndpoint)
+ [Obsolete("This constructor will be removed in the 0.13 version", false)]
+ public NameClient(IPEndPoint remoteEndpoint)
{
Initialize(remoteEndpoint);
+ _cache = TangFactory.GetTang().NewInjector().GetInstance<NameCache>();
_disposed = false;
}
@@ -123,6 +147,31 @@ namespace Org.Apache.REEF.Network.Naming
/// <summary>
/// Synchronously looks up the IPEndpoint for the registered identifier.
+ /// Uses cache if it has entry
+ /// </summary>
+ /// <param name="id">The identifier to look up</param>
+ /// <returns>The mapped IPEndpoint for the identifier, or null if
+ /// the identifier has not been registered with the NameService</returns>
+ public IPEndPoint CacheLookup(string id)
+ {
+ if (id == null)
+ {
+ Exceptions.Throw(new ArgumentNullException("id"), _logger);
+ }
+
+ IPEndPoint value = _cache.Get(id);
+
+ if (value != null)
+ {
+ return value;
+ }
+
+ return Lookup(id);
+ }
+
+ /// <summary>
+ /// Synchronously looks up the IPEndpoint for the registered identifier.
+ /// Does not use cache
/// </summary>
/// <param name="id">The identifier to look up</param>
/// <returns>The mapped IPEndpoint for the identifier, or null if
@@ -137,6 +186,7 @@ namespace Org.Apache.REEF.Network.Naming
List<NameAssignment> assignments = Lookup(new List<string> { id });
if (assignments != null && assignments.Count > 0)
{
+ _cache.Set(id, assignments.First().Endpoint);
return assignments.First().Endpoint;
}
@@ -145,6 +195,7 @@ namespace Org.Apache.REEF.Network.Naming
/// <summary>
/// Synchronously looks up the IPEndpoint for each of the registered identifiers in the list.
+ /// Do not use cache
/// </summary>
/// <param name="ids">The list of identifiers to look up</param>
/// <returns>The list of NameAssignments representing a pair of identifer
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network/Naming/Parameters/NameCacheConfiguration.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Naming/Parameters/NameCacheConfiguration.cs b/lang/cs/Org.Apache.REEF.Network/Naming/Parameters/NameCacheConfiguration.cs
new file mode 100644
index 0000000..62bdab9
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Naming/Parameters/NameCacheConfiguration.cs
@@ -0,0 +1,41 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Naming.Parameters
+{
+ public static class NameCacheConfiguration
+ {
+ [NamedParameter("Time interval in milliseconds before the cache entry expires", "timeinterval", "300000")]
+ public class CacheEntryExpiryTime : Name<double>
+ {
+ }
+
+ [NamedParameter("Maximum cache memory in MB", "cachememorylimit", "20")]
+ public class CacheMemoryLimit : Name<string>
+ {
+ }
+
+ [NamedParameter("Polling interval for checking cache memory", "cachepollinginterval", "00:20:00")]
+ public class PollingInterval : Name<string>
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network/NetworkService/NsConnection.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/NetworkService/NsConnection.cs b/lang/cs/Org.Apache.REEF.Network/NetworkService/NsConnection.cs
index 1cb47d3..cdf9c56 100644
--- a/lang/cs/Org.Apache.REEF.Network/NetworkService/NsConnection.cs
+++ b/lang/cs/Org.Apache.REEF.Network/NetworkService/NsConnection.cs
@@ -76,7 +76,7 @@ namespace Org.Apache.REEF.Network.NetworkService
string destStr = _destId.ToString();
LOGGER.Log(Level.Verbose, "Network service opening connection to {0}...", destStr);
- IPEndPoint destAddr = _nameClient.Lookup(_destId.ToString());
+ IPEndPoint destAddr = _nameClient.CacheLookup(_destId.ToString());
if (destAddr == null)
{
throw new RemotingException("Cannot register Identifier with NameService");
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/e5aef161/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj b/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
index a0e6038..2d62cb4 100644
--- a/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
+++ b/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
@@ -46,6 +46,7 @@ under the License.
</Reference>
<Reference Include="System" />
<Reference Include="System.Core" />
+ <Reference Include="System.Runtime.Caching" />
<Reference Include="System.Runtime.Serialization" />
<Reference Include="System.Xml" />
</ItemGroup>
@@ -129,6 +130,8 @@ under the License.
<Compile Include="Naming\Events\NamingUnregisterRequest.cs" />
<Compile Include="Naming\Events\NamingUnregisterResponse.cs" />
<Compile Include="Naming\INameServer.cs" />
+ <Compile Include="Naming\NameCache.cs" />
+ <Compile Include="Naming\Parameters\NameCacheConfiguration.cs" />
<Compile Include="Naming\NameClient.cs" />
<Compile Include="Naming\NameLookupClient.cs" />
<Compile Include="Naming\NameRegisterClient.cs" />