diff options
Diffstat (limited to 'java/jvpp-stats/io/fd/jvpp/stats/future')
5 files changed, 380 insertions, 0 deletions
diff --git a/java/jvpp-stats/io/fd/jvpp/stats/future/AbstractFutureJVppInvoker.java b/java/jvpp-stats/io/fd/jvpp/stats/future/AbstractFutureJVppInvoker.java new file mode 100644 index 0000000..e66cfd8 --- /dev/null +++ b/java/jvpp-stats/io/fd/jvpp/stats/future/AbstractFutureJVppInvoker.java @@ -0,0 +1,170 @@ +/* + * Copyright (c) 2019 Cisco and/or its affiliates. + * + * Licensed 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 io.fd.jvpp.stats.future; + + +import io.fd.jvpp.JVpp; +import io.fd.jvpp.JVppRegistry; +import io.fd.jvpp.VppInvocationException; +import io.fd.jvpp.dto.JVppDump; +import io.fd.jvpp.dto.JVppReply; +import io.fd.jvpp.dto.JVppReplyDump; +import io.fd.jvpp.dto.JVppRequest; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +/** + * Future facade on top of JVpp + */ +public abstract class AbstractFutureJVppInvoker implements FutureJVppInvoker { + + private final JVpp jvpp; + private final JVppRegistry registry; + + /** + * Guarded by self + */ + private final Map<Integer, CompletableFuture<? extends JVppReply<?>>> requests; + + protected AbstractFutureJVppInvoker(final JVpp jvpp, final JVppRegistry registry, + final Map<Integer, CompletableFuture<? extends JVppReply<?>>> requestMap) { + this.jvpp = Objects.requireNonNull(jvpp, "jvpp should not be null"); + this.registry = Objects.requireNonNull(registry, "registry should not be null"); + // Request map represents the shared state between this facade and it's callback + // where facade puts futures in and callback completes + removes them + this.requests = Objects.requireNonNull(requestMap, "Null requestMap"); + } + + protected final Map<Integer, CompletableFuture<? extends JVppReply<?>>> getRequests() { + synchronized (requests) { + return requests; + } + } + + // TODO use Optional in Future, java8 + + @Override + @SuppressWarnings("unchecked") + public <REQ extends JVppRequest, REPLY extends JVppReply<REQ>> CompletionStage<REPLY> send(REQ req) { + try { + // jvpp.send() can go to waiting state if sending queue is full, putting it into same + // synchronization block as used by receiving part (synchronized(requests)) can lead + // to deadlock between these two sides or at least slowing sending process by slow + // reader + final CompletableFuture<REPLY> replyCompletableFuture; + final int contextId = jvpp.send(req); + + if(req instanceof JVppDump) { + throw new IllegalArgumentException("Send with empty reply dump has to be used in case of dump calls"); + } + + synchronized(requests) { + CompletableFuture<? extends JVppReply<?>> replyFuture = requests.get(contextId); + if (replyFuture == null) { + // reply not yet received, put new future into map + replyCompletableFuture = new CompletableFuture<>(); + requests.put(contextId, replyCompletableFuture); + } else { + // reply already received (should be completed by reader), + // remove future from map and return it to caller + replyCompletableFuture = (CompletableFuture<REPLY>) replyFuture; + requests.remove(contextId); + } + } + + // TODO in case of timeouts/missing replies, requests from the map are not removed + // consider adding cancel method, that would remove requests from the map and cancel + // associated replyCompletableFuture + + return replyCompletableFuture; + } catch (VppInvocationException ex) { + final CompletableFuture<REPLY> replyCompletableFuture = new CompletableFuture<>(); + replyCompletableFuture.completeExceptionally(ex); + return replyCompletableFuture; + } + } + + @Override + @SuppressWarnings("unchecked") + public <REQ extends JVppRequest, REPLY extends JVppReply<REQ>, DUMP extends JVppReplyDump<REQ, REPLY>> CompletionStage<DUMP> send( + REQ req, DUMP emptyReplyDump) { + try { + // jvpp.send() can go to waiting state if sending queue is full, + // putting it into same synchronization block as used by receiving part (synchronized(requests)) + // can lead to deadlock between these two sides or at least slowing sending process by slow reader + final CompletableDumpFuture<DUMP> replyDumpFuture; + final int contextId = jvpp.send(req); + + if(!(req instanceof JVppDump)) { + throw new IllegalArgumentException("Send without empty reply dump has to be used in case of regular calls"); + } + + synchronized(requests) { + CompletableFuture<? extends JVppReply<?>> replyFuture = requests.get(contextId); + if (replyFuture == null) { + // reply not received yet, put new future to map + replyDumpFuture = new CompletableDumpFuture<>(contextId, emptyReplyDump); + requests.put(contextId, replyDumpFuture); + } else { + // reply already received, save existing future + replyDumpFuture = (CompletableDumpFuture<DUMP>) replyFuture; + } + } + + synchronized (requests) { + // reply already received, complete future + replyDumpFuture.complete(replyDumpFuture.getReplyDump()); + requests.remove(contextId); + } + + // TODO in case of timeouts/missing replies, requests from the map are not removed + // consider adding cancel method, that would remove requests from the map and cancel + // associated replyCompletableFuture + + return replyDumpFuture; + } catch (VppInvocationException ex) { + final CompletableFuture<DUMP> replyCompletableFuture = new CompletableFuture<>(); + replyCompletableFuture.completeExceptionally(ex); + return replyCompletableFuture; + } + } + + public static final class CompletableDumpFuture<T extends JVppReplyDump<?, ?>> extends CompletableFuture<T> { + private final T replyDump; + private final int contextId; + + public CompletableDumpFuture(final int contextId, final T emptyDump) { + this.contextId = contextId; + this.replyDump = emptyDump; + } + + public int getContextId() { + return contextId; + } + + public T getReplyDump() { + return replyDump; + } + } + + @Override + public void close() throws Exception { + jvpp.close(); + } +} diff --git a/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppInvoker.java b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppInvoker.java new file mode 100644 index 0000000..c93c332 --- /dev/null +++ b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppInvoker.java @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2019 Cisco and/or its affiliates. + * + * Licensed 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 io.fd.jvpp.stats.future; + + +import io.fd.jvpp.dto.JVppReply; +import io.fd.jvpp.dto.JVppReplyDump; +import io.fd.jvpp.dto.JVppRequest; +import io.fd.jvpp.notification.EventRegistryProvider; +import java.util.concurrent.CompletionStage; + +/** +* Future facade on top of JVpp +*/ +public interface FutureJVppInvoker extends EventRegistryProvider, AutoCloseable { + + /** + * Invoke asynchronous operation on VPP + * + * @return CompletionStage with future result of an async VPP call + * @throws io.fd.jvpp.VppInvocationException when send request failed with details + */ + <REQ extends JVppRequest, REPLY extends JVppReply<REQ>> CompletionStage<REPLY> send(REQ req); + + + /** + * Invoke asynchronous dump operation on VPP + * + * @return CompletionStage with aggregated future result of an async VPP dump call + * @throws io.fd.jvpp.VppInvocationException when send request failed with details + */ + <REQ extends JVppRequest, REPLY extends JVppReply<REQ>, DUMP extends JVppReplyDump<REQ, REPLY>> CompletionStage<DUMP> send( + REQ req, DUMP emptyReplyDump); +} diff --git a/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStats.java b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStats.java new file mode 100644 index 0000000..288b556 --- /dev/null +++ b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStats.java @@ -0,0 +1,28 @@ +/* + * Copyright (c) 2019 Cisco and/or its affiliates. + * + * Licensed 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 io.fd.jvpp.stats.future; + +import io.fd.jvpp.stats.dto.InterfaceStatisticsDetailsReplyDump; + +/** + * <p>Async facade extension adding specific methods for each request invocation + */ +public interface FutureJVppStats extends io.fd.jvpp.stats.future.FutureJVppInvoker { + + java.util.concurrent.CompletionStage<InterfaceStatisticsDetailsReplyDump> interfaceStatisticsDump( + io.fd.jvpp.stats.dto.InterfaceStatisticsDump request); +} diff --git a/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacade.java b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacade.java new file mode 100644 index 0000000..470a2fd --- /dev/null +++ b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacade.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2019 Cisco and/or its affiliates. + * + * Licensed 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 io.fd.jvpp.stats.future; + +import io.fd.jvpp.notification.EventRegistry; +import io.fd.jvpp.stats.dto.InterfaceStatisticsDetailsReplyDump; + +/** + * <p>Implementation of FutureJVpp based on AbstractFutureJVppInvoker + */ +public class FutureJVppStatsFacade extends io.fd.jvpp.stats.future.AbstractFutureJVppInvoker + implements FutureJVppStats { + + /** + * <p>Create FutureJVppStatsFacade object for provided JVpp instance. + * Constructor internally creates FutureJVppFacadeCallback class for processing callbacks + * and then connects to provided JVpp instance + * + * @param jvpp provided io.fd.jvpp.JVpp instance + * @throws java.io.IOException in case instance cannot connect to JVPP + */ + public FutureJVppStatsFacade(final io.fd.jvpp.JVppRegistry registry, final io.fd.jvpp.JVpp jvpp) + throws java.io.IOException { + super(jvpp, registry, new java.util.HashMap<>()); + java.util.Objects.requireNonNull(registry, "JVppRegistry should not be null"); + registry.register(jvpp, new FutureJVppStatsFacadeCallback(getRequests())); + } + + @Override + public java.util.concurrent.CompletionStage<InterfaceStatisticsDetailsReplyDump> interfaceStatisticsDump( + io.fd.jvpp.stats.dto.InterfaceStatisticsDump request) { + return send(request, new InterfaceStatisticsDetailsReplyDump()); + } + + @Override + public EventRegistry getEventRegistry() { + return null; + } +} diff --git a/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacadeCallback.java b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacadeCallback.java new file mode 100644 index 0000000..a215b89 --- /dev/null +++ b/java/jvpp-stats/io/fd/jvpp/stats/future/FutureJVppStatsFacadeCallback.java @@ -0,0 +1,81 @@ +/* + * Copyright (c) 2019 Cisco and/or its affiliates. + * + * Licensed 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 io.fd.jvpp.stats.future; + +import io.fd.jvpp.stats.dto.InterfaceStatisticsDetailsReplyDump; + +/** + * <p>Async facade callback setting values to future objects + */ +public final class FutureJVppStatsFacadeCallback implements io.fd.jvpp.stats.callback.JVppStatsGlobalCallback { + + private static final java.util.logging.Logger LOG = java.util.logging.Logger.getLogger( + FutureJVppStatsFacadeCallback.class.getName()); + private final java.util.Map<Integer, java.util.concurrent.CompletableFuture<? extends io.fd.jvpp.dto.JVppReply<?>>> + requests; + + public FutureJVppStatsFacadeCallback( + final java.util.Map<Integer, java.util.concurrent.CompletableFuture<? extends io.fd.jvpp.dto.JVppReply<?>>> requestMap) { + this.requests = requestMap; + } + + @Override + @SuppressWarnings("unchecked") + public void onError(io.fd.jvpp.VppCallbackException reply) { + final java.util.concurrent.CompletableFuture<io.fd.jvpp.dto.JVppReply<?>> completableFuture; + + synchronized (requests) { + completableFuture = (java.util.concurrent.CompletableFuture<io.fd.jvpp.dto.JVppReply<?>>) requests + .get(reply.getCtxId()); + } + + if (completableFuture != null) { + completableFuture.completeExceptionally(reply); + + synchronized (requests) { + requests.remove(reply.getCtxId()); + } + } + } + + @Override + @SuppressWarnings("unchecked") + public void onInterfaceStatisticsDetails(final io.fd.jvpp.stats.dto.InterfaceStatisticsDetails reply) { + io.fd.jvpp.stats.future.AbstractFutureJVppInvoker.CompletableDumpFuture<InterfaceStatisticsDetailsReplyDump> + completableFuture; + final int replyId = reply.context; + if (LOG.isLoggable(java.util.logging.Level.FINE)) { + LOG.fine(String.format("Received InterfaceStatisticsDetails event message: %s", reply)); + } + synchronized (requests) { + completableFuture = + (io.fd.jvpp.stats.future.AbstractFutureJVppInvoker.CompletableDumpFuture<InterfaceStatisticsDetailsReplyDump>) requests + .get(replyId); + + if (completableFuture == null) { + // reply received before writer created future, + // create new future, and put into map to notify sender that reply is already received, + // following details replies will add information to this future + completableFuture = + new io.fd.jvpp.stats.future.AbstractFutureJVppInvoker.CompletableDumpFuture<>(replyId, + new InterfaceStatisticsDetailsReplyDump()); + requests.put(replyId, completableFuture); + } + completableFuture.getReplyDump().interfaceStatisticsDetails = reply; + } + } +} |