| /* |
| * 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.igfs; |
| |
| import java.util.Collection; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.LinkedBlockingDeque; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.locks.Condition; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReentrantLock; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.internal.IgniteInterruptedCheckedException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.managers.communication.GridMessageListener; |
| import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; |
| import org.apache.ignite.internal.processors.igfs.meta.IgfsMetaFileRangeDeleteProcessor; |
| import org.apache.ignite.internal.processors.igfs.meta.IgfsMetaFileRangeUpdateProcessor; |
| import org.apache.ignite.internal.util.GridConcurrentHashSet; |
| import org.apache.ignite.internal.util.GridSpinReadWriteLock; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.P1; |
| import org.apache.ignite.internal.util.typedef.X; |
| import org.apache.ignite.internal.util.typedef.internal.LT; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.internal.util.worker.GridWorker; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.apache.ignite.thread.IgniteThread; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; |
| import static org.apache.ignite.internal.GridTopic.TOPIC_IGFS; |
| import static org.apache.ignite.internal.managers.communication.GridIoPolicy.IGFS_POOL; |
| import static org.apache.ignite.internal.processors.igfs.IgfsFileAffinityRange.RANGE_STATUS_INITIAL; |
| import static org.apache.ignite.internal.processors.igfs.IgfsFileAffinityRange.RANGE_STATUS_MOVED; |
| import static org.apache.ignite.internal.processors.igfs.IgfsFileAffinityRange.RANGE_STATUS_MOVING; |
| |
| /** |
| * IGFS fragmentizer manager. |
| */ |
| public class IgfsFragmentizerManager extends IgfsManager { |
| /** Message offer wait interval. */ |
| private static final int MSG_OFFER_TIMEOUT = 1000; |
| |
| /** Fragmentizer files check interval. */ |
| private static final int FRAGMENTIZER_CHECK_INTERVAL = 3000; |
| |
| /** Message send retry interval. */ |
| private static final int MESSAGE_SEND_RETRY_INTERVAL = 1000; |
| |
| /** How many times retry message send. */ |
| private static final int MESSAGE_SEND_RETRY_COUNT = 3; |
| |
| /** Manager stopping flag. */ |
| private volatile boolean stopping; |
| |
| /** Coordinator worker. */ |
| private volatile FragmentizerCoordinator fragmentizerCrd; |
| |
| /** This variable is used in tests only. */ |
| @SuppressWarnings("FieldCanBeLocal") |
| private volatile boolean fragmentizerEnabled = true; |
| |
| /** Fragmentizer worker. */ |
| private FragmentizerWorker fragmentizerWorker; |
| |
| /** Shutdown lock. */ |
| private GridSpinReadWriteLock rw = new GridSpinReadWriteLock(); |
| |
| /** Message topic. */ |
| private Object topic; |
| |
| /** {@inheritDoc} */ |
| @Override protected void start0() throws IgniteCheckedException { |
| if (!igfsCtx.configuration().isFragmentizerEnabled()) |
| return; |
| |
| // We care only about node leave and fail events. |
| igfsCtx.kernalContext().event().addLocalEventListener(new GridLocalEventListener() { |
| @Override public void onEvent(Event evt) { |
| assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED; |
| |
| DiscoveryEvent discoEvt = (DiscoveryEvent)evt; |
| |
| checkLaunchCoordinator(discoEvt); |
| } |
| }, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| fragmentizerWorker = new FragmentizerWorker(); |
| |
| String igfsName = igfsCtx.configuration().getName(); |
| |
| topic = F.isEmpty(igfsName) ? TOPIC_IGFS : TOPIC_IGFS.topic(igfsName); |
| |
| igfsCtx.kernalContext().io().addMessageListener(topic, fragmentizerWorker); |
| |
| new IgniteThread(fragmentizerWorker).start(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void onKernalStart0() throws IgniteCheckedException { |
| if (igfsCtx.configuration().isFragmentizerEnabled()) { |
| // Check at startup if this node is a fragmentizer coordinator. |
| DiscoveryEvent locJoinEvt = igfsCtx.kernalContext().discovery().localJoinEvent(); |
| |
| checkLaunchCoordinator(locJoinEvt); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("BusyWait") |
| @Override protected void onKernalStop0(boolean cancel) { |
| boolean interrupted = false; |
| |
| // Busy wait is intentional. |
| while (true) { |
| try { |
| if (rw.tryWriteLock(200, TimeUnit.MILLISECONDS)) |
| break; |
| else |
| Thread.sleep(200); |
| } |
| catch (InterruptedException ignore) { |
| // Preserve interrupt status & ignore. |
| // Note that interrupted flag is cleared. |
| interrupted = true; |
| } |
| } |
| |
| try { |
| if (interrupted) |
| Thread.currentThread().interrupt(); |
| |
| stopping = true; |
| } |
| finally { |
| rw.writeUnlock(); |
| } |
| |
| synchronized (this) { |
| if (fragmentizerCrd != null) |
| fragmentizerCrd.cancel(); |
| } |
| |
| if (fragmentizerWorker != null) |
| fragmentizerWorker.cancel(); |
| |
| U.join(fragmentizerCrd, log); |
| U.join(fragmentizerWorker, log); |
| } |
| |
| /** |
| * @param nodeId Node ID to send message to. |
| * @param msg Message to send. |
| * @throws IgniteCheckedException If send failed. |
| */ |
| private void sendWithRetries(UUID nodeId, IgfsCommunicationMessage msg) throws IgniteCheckedException { |
| for (int i = 0; i < MESSAGE_SEND_RETRY_COUNT; i++) { |
| try { |
| igfsCtx.send(nodeId, topic, msg, IGFS_POOL); |
| |
| return; |
| } |
| catch (IgniteCheckedException e) { |
| if (!igfsCtx.kernalContext().discovery().alive(nodeId)) |
| throw new ClusterTopologyCheckedException("Failed to send message (node left the grid) " + |
| "[nodeId=" + nodeId + ", msg=" + msg + ']'); |
| |
| if (i == MESSAGE_SEND_RETRY_COUNT - 1) |
| throw e; |
| |
| U.sleep(MESSAGE_SEND_RETRY_INTERVAL); |
| } |
| } |
| } |
| |
| /** |
| * Checks if current node is the oldest node in topology and starts coordinator thread if so. |
| * Note that once node is the oldest one, it will be the oldest until it leaves grid. |
| * |
| * @param discoEvt Discovery event. |
| */ |
| private void checkLaunchCoordinator(DiscoveryEvent discoEvt) { |
| rw.readLock(); |
| |
| try { |
| if (stopping) |
| return; |
| |
| if (fragmentizerCrd == null) { |
| long minNodeOrder = Long.MAX_VALUE; |
| |
| Collection<ClusterNode> nodes = discoEvt.topologyNodes(); |
| |
| for (ClusterNode node : nodes) { |
| if (node.order() < minNodeOrder && igfsCtx.igfsNode(node)) |
| minNodeOrder = node.order(); |
| } |
| |
| ClusterNode locNode = igfsCtx.kernalContext().grid().localNode(); |
| |
| if (locNode.order() == minNodeOrder) { |
| if (log.isDebugEnabled()) |
| log.debug("Detected local node to be the eldest IGFS node in topology, starting fragmentizer " + |
| "coordinator thread [discoEvt=" + discoEvt + ", locNode=" + locNode + ']'); |
| |
| synchronized (this) { |
| if (fragmentizerCrd == null && !stopping) { |
| fragmentizerCrd = new FragmentizerCoordinator(); |
| |
| new IgniteThread(fragmentizerCrd).start(); |
| } |
| } |
| } |
| } |
| } |
| finally { |
| rw.readUnlock(); |
| } |
| } |
| |
| /** |
| * Processes fragmentizer request. For each range assigned to this node: |
| * <ul> |
| * <li>Mark range as moving indicating that block copying started.</li> |
| * <li>Copy blocks to non-colocated keys.</li> |
| * <li>Update map to indicate that blocks were copied and old blocks should be deleted.</li> |
| * <li>Delete old blocks.</li> |
| * <li>Remove range from file map.</li> |
| * </ul> |
| * |
| * @param req Request. |
| * @throws IgniteCheckedException In case of error. |
| */ |
| @SuppressWarnings("fallthrough") |
| private void processFragmentizerRequest(IgfsFragmentizerRequest req) throws IgniteCheckedException { |
| req.finishUnmarshal(igfsCtx.kernalContext().config().getMarshaller(), null); |
| |
| Collection<IgfsFileAffinityRange> ranges = req.fragmentRanges(); |
| IgniteUuid fileId = req.fileId(); |
| |
| IgfsEntryInfo fileInfo = igfsCtx.meta().info(fileId); |
| |
| if (fileInfo == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to find file info for fragmentizer request: " + req); |
| |
| return; |
| } |
| |
| if (log.isDebugEnabled()) |
| log.debug("Moving file ranges for fragmentizer request [req=" + req + ", fileInfo=" + fileInfo + ']'); |
| |
| for (IgfsFileAffinityRange range : ranges) { |
| try { |
| IgfsEntryInfo updated; |
| |
| switch (range.status()) { |
| case RANGE_STATUS_INITIAL: { |
| // Mark range as moving. |
| updated = igfsCtx.meta().updateInfo( |
| fileId, new IgfsMetaFileRangeUpdateProcessor(range, RANGE_STATUS_MOVING)); |
| |
| if (updated == null) { |
| igfsCtx.data().cleanBlocks(fileInfo, range, true); |
| |
| continue; |
| } |
| |
| // Fall-through. |
| } |
| |
| case RANGE_STATUS_MOVING: { |
| // Move colocated blocks. |
| igfsCtx.data().spreadBlocks(fileInfo, range); |
| |
| // Mark range as moved. |
| updated = igfsCtx.meta().updateInfo( |
| fileId, new IgfsMetaFileRangeUpdateProcessor(range, RANGE_STATUS_MOVED)); |
| |
| if (updated == null) { |
| igfsCtx.data().cleanBlocks(fileInfo, range, true); |
| |
| continue; |
| } |
| |
| // Fall-through. |
| } |
| |
| case RANGE_STATUS_MOVED: { |
| // Remove old blocks. |
| igfsCtx.data().cleanBlocks(fileInfo, range, false); |
| |
| // Remove range from map. |
| updated = igfsCtx.meta().updateInfo(fileId, new IgfsMetaFileRangeDeleteProcessor(range)); |
| |
| if (updated == null) |
| igfsCtx.data().cleanBlocks(fileInfo, range, true); |
| } |
| } |
| } |
| catch (IgfsInvalidRangeException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to update file range " + |
| "[range=" + range + "fileId=" + fileId + ", err=" + e.getMessage() + ']'); |
| } |
| } |
| } |
| |
| /** |
| * Fragmentizer coordinator thread. |
| */ |
| private class FragmentizerCoordinator extends GridWorker implements GridLocalEventListener, GridMessageListener { |
| /** Files being fragmented. */ |
| private ConcurrentMap<IgniteUuid, Collection<UUID>> fragmentingFiles = new ConcurrentHashMap<>(); |
| |
| /** Node IDs captured on start. */ |
| private volatile Collection<UUID> startSync; |
| |
| /** Wait lock. */ |
| private Lock lock = new ReentrantLock(); |
| |
| /** Wait condition. */ |
| private Condition cond = lock.newCondition(); |
| |
| /** |
| * Constructor. |
| */ |
| protected FragmentizerCoordinator() { |
| super(igfsCtx.kernalContext().igniteInstanceName(), "fragmentizer-coordinator", |
| igfsCtx.kernalContext().log(IgfsFragmentizerManager.class)); |
| |
| igfsCtx.kernalContext().event().addLocalEventListener(this, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| igfsCtx.kernalContext().io().addMessageListener(topic, this); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException { |
| // Wait for all previous fragmentizer tasks to complete. |
| syncStart(); |
| |
| while (!isCancelled()) { |
| // If we have room for files, add them to fragmentizer. |
| try { |
| while (fragmentingFiles.size() < igfsCtx.configuration().getFragmentizerConcurrentFiles()) { |
| IgfsEntryInfo fileInfo = fileForFragmentizer(fragmentingFiles.keySet()); |
| |
| // If no colocated files found, exit loop. |
| if (fileInfo == null) |
| break; |
| |
| requestFragmenting(fileInfo); |
| } |
| } |
| catch (IgniteCheckedException | IgniteException e) { |
| if (!X.hasCause(e, InterruptedException.class) && !X.hasCause(e, IgniteInterruptedCheckedException.class)) |
| LT.error(log, e, "Failed to get fragmentizer file info (will retry)."); |
| else { |
| if (log.isDebugEnabled()) |
| log.debug("Got interrupted exception in fragmentizer coordinator (grid is stopping)."); |
| |
| break; // While. |
| } |
| } |
| |
| lock.lock(); |
| |
| try { |
| cond.await(FRAGMENTIZER_CHECK_INTERVAL, MILLISECONDS); |
| } |
| finally { |
| lock.unlock(); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onEvent(Event evt) { |
| assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED; |
| |
| DiscoveryEvent discoEvt = (DiscoveryEvent)evt; |
| |
| if (log.isDebugEnabled()) |
| log.debug("Processing node leave event: " + discoEvt); |
| |
| boolean signal = false; |
| |
| Collection<UUID> startSync0 = startSync; |
| |
| if (startSync0 != null && !startSync0.isEmpty()) { |
| startSync0.remove(discoEvt.eventNode().id()); |
| |
| if (startSync0.isEmpty()) { |
| if (log.isDebugEnabled()) |
| log.debug("Completed fragmentizer coordinator sync start."); |
| |
| signal = true; |
| } |
| } |
| |
| if (!signal) { |
| Iterator<Map.Entry<IgniteUuid, Collection<UUID>>> it = fragmentingFiles.entrySet().iterator(); |
| |
| while (it.hasNext()) { |
| Map.Entry<IgniteUuid, Collection<UUID>> entry = it.next(); |
| |
| Collection<UUID> nodeIds = entry.getValue(); |
| |
| if (nodeIds.remove(discoEvt.eventNode().id())) { |
| if (nodeIds.isEmpty()) { |
| if (log.isDebugEnabled()) |
| log.debug("Received all responses for fragmentizer task [fileId=" + entry.getKey() + |
| ']'); |
| |
| it.remove(); |
| |
| signal = true; |
| } |
| } |
| } |
| } |
| |
| if (signal) |
| wakeUp(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onMessage(UUID nodeId, Object msg, byte plc) { |
| if (msg instanceof IgfsFragmentizerResponse) { |
| IgfsFragmentizerResponse res = (IgfsFragmentizerResponse)msg; |
| |
| IgniteUuid fileId = res.fileId(); |
| |
| Collection<UUID> nodeIds = fragmentingFiles.get(fileId); |
| |
| if (nodeIds != null) { |
| if (nodeIds.remove(nodeId)) { |
| if (nodeIds.isEmpty()) { |
| if (log.isDebugEnabled()) |
| log.debug("Received all responses for fragmentizer task [fileId=" + fileId + ']'); |
| |
| fragmentingFiles.remove(fileId, nodeIds); |
| |
| wakeUp(); |
| } |
| } |
| } |
| else |
| log.warning("Received fragmentizer response for file ID which was not requested (will ignore) " + |
| "[nodeId=" + nodeId + ", fileId=" + res.fileId() + ']'); |
| } |
| else if (msg instanceof IgfsSyncMessage) { |
| IgfsSyncMessage sync = (IgfsSyncMessage)msg; |
| |
| if (sync.response() && sync.order() == igfsCtx.kernalContext().grid().localNode().order()) { |
| if (log.isDebugEnabled()) |
| log.debug("Received fragmentizer sync response from remote node: " + nodeId); |
| |
| Collection<UUID> startSync0 = startSync; |
| |
| if (startSync0 != null) { |
| startSync0.remove(nodeId); |
| |
| if (startSync0.isEmpty()) { |
| if (log.isDebugEnabled()) |
| log.debug("Completed fragmentizer coordinator sync start: " + startSync0); |
| |
| wakeUp(); |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Signals condition. |
| */ |
| private void wakeUp() { |
| lock.lock(); |
| |
| try { |
| cond.signalAll(); |
| } |
| finally { |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * Sends sync message to remote nodes and awaits for response from all nodes. |
| * |
| * @throws InterruptedException If waiting was interrupted. |
| */ |
| private void syncStart() throws InterruptedException { |
| Collection<UUID> startSync0 = startSync = new GridConcurrentHashSet<>( |
| F.viewReadOnly( |
| igfsCtx.kernalContext().discovery().allNodes(), |
| F.node2id(), |
| new P1<ClusterNode>() { |
| @Override public boolean apply(ClusterNode n) { |
| return igfsCtx.igfsNode(n); |
| } |
| })); |
| |
| ClusterNode locNode = igfsCtx.kernalContext().grid().localNode(); |
| |
| while (!startSync0.isEmpty()) { |
| for (UUID nodeId : startSync0) { |
| IgfsSyncMessage syncReq = new IgfsSyncMessage(locNode.order(), false); |
| |
| try { |
| if (log.isDebugEnabled()) |
| log.debug("Sending fragmentizer sync start request to remote node [nodeId=" + nodeId + |
| ", syncReq=" + syncReq + ']'); |
| |
| sendWithRetries(nodeId, syncReq); |
| |
| // Close window between message sending and discovery event. |
| if (!igfsCtx.kernalContext().discovery().alive(nodeId)) |
| startSync0.remove(nodeId); |
| } |
| catch (IgniteCheckedException e) { |
| if (e.hasCause(ClusterTopologyCheckedException.class)) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send sync message to remote node (node has left the grid): " + |
| nodeId); |
| } |
| else |
| U.error(log, "Failed to send synchronize message to remote node (will not wait for reply): " + |
| nodeId, e); |
| |
| startSync0.remove(nodeId); |
| } |
| } |
| |
| lock.lock(); |
| |
| try { |
| if (!startSync0.isEmpty()) |
| cond.await(10000, MILLISECONDS); |
| } |
| finally { |
| lock.unlock(); |
| } |
| } |
| } |
| |
| /** |
| * Starts file fragmenting. Will group file affinity ranges by nodes and send requests to each node. |
| * File will be considered processed when each node replied with success (or error) or left the grid. |
| * |
| * @param fileInfo File info to process. |
| */ |
| private void requestFragmenting(IgfsEntryInfo fileInfo) { |
| IgfsFileMap map = fileInfo.fileMap(); |
| |
| assert map != null && !map.ranges().isEmpty(); |
| |
| Map<UUID, Collection<IgfsFileAffinityRange>> grpMap = U.newHashMap(map.ranges().size()); |
| |
| for (IgfsFileAffinityRange range : map.ranges()) { |
| UUID nodeId = igfsCtx.data().affinityNode(range.affinityKey()).id(); |
| |
| Collection<IgfsFileAffinityRange> nodeRanges = grpMap.get(nodeId); |
| |
| if (nodeRanges == null) { |
| nodeRanges = new LinkedList<>(); |
| |
| grpMap.put(nodeId, nodeRanges); |
| } |
| |
| nodeRanges.addAll(range.split(igfsCtx.data().groupBlockSize())); |
| } |
| |
| Collection<UUID> nodeIds = new IdentityHashSet(grpMap.keySet()); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Calculating fragmentizer groups for file [fileInfo=" + fileInfo + |
| ", nodeIds=" + nodeIds + ']'); |
| |
| // Put assignment to map first. |
| Object old = fragmentingFiles.putIfAbsent(fileInfo.id(), nodeIds); |
| |
| assert old == null; |
| |
| for (Map.Entry<UUID, Collection<IgfsFileAffinityRange>> entry : grpMap.entrySet()) { |
| UUID nodeId = entry.getKey(); |
| |
| IgfsFragmentizerRequest msg = new IgfsFragmentizerRequest(fileInfo.id(), entry.getValue()); |
| |
| try { |
| if (log.isDebugEnabled()) |
| log.debug("Sending fragmentizer request to remote node [nodeId=" + nodeId + |
| ", fileId=" + fileInfo.id() + ", msg=" + msg + ']'); |
| |
| sendWithRetries(nodeId, msg); |
| } |
| catch (IgniteCheckedException e) { |
| if (e.hasCause(ClusterTopologyCheckedException.class)) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send fragmentizer request to remote node (node left grid): " + |
| nodeId); |
| } |
| else |
| U.error(log, "Failed to send fragmentizer request to remote node [nodeId=" + nodeId + |
| ", msg=" + msg + ']', e); |
| |
| nodeIds.remove(nodeId); |
| } |
| } |
| |
| if (nodeIds.isEmpty()) { |
| if (log.isDebugEnabled()) |
| log.debug("Got empty wait set for fragmentized file: " + fileInfo); |
| |
| fragmentingFiles.remove(fileInfo.id(), nodeIds); |
| } |
| } |
| } |
| |
| /** |
| * Gets next file for fragmentizer to be processed. |
| * |
| * @param exclude File IDs to exclude (the ones that are currently being processed). |
| * @return File ID to process or {@code null} if there are no such files. |
| * @throws IgniteCheckedException In case of error. |
| */ |
| @Nullable private IgfsEntryInfo fileForFragmentizer(Collection<IgniteUuid> exclude) throws IgniteCheckedException { |
| return fragmentizerEnabled ? igfsCtx.meta().fileForFragmentizer(exclude) : null; |
| } |
| |
| /** |
| * Fragmentizer worker thread. |
| */ |
| private class FragmentizerWorker extends GridWorker implements GridMessageListener { |
| /** Requests for this worker. */ |
| private BlockingQueue<IgniteBiTuple<UUID, IgfsCommunicationMessage>> msgs = new LinkedBlockingDeque<>(); |
| |
| /** |
| * Constructor. |
| */ |
| protected FragmentizerWorker() { |
| super(igfsCtx.kernalContext().igniteInstanceName(), "fragmentizer-worker", |
| igfsCtx.kernalContext().log(IgfsFragmentizerManager.class)); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onMessage(UUID nodeId, Object msg, byte plc) { |
| if (msg instanceof IgfsFragmentizerRequest || |
| msg instanceof IgfsSyncMessage) { |
| if (log.isDebugEnabled()) |
| log.debug("Received fragmentizer request from remote node [nodeId=" + nodeId + |
| ", msg=" + msg + ']'); |
| |
| IgniteBiTuple<UUID, IgfsCommunicationMessage> tup = F.t(nodeId, (IgfsCommunicationMessage)msg); |
| |
| try { |
| if (!msgs.offer(tup, MSG_OFFER_TIMEOUT, TimeUnit.MILLISECONDS)) { |
| U.error(log, "Failed to process fragmentizer communication message (will discard) " + |
| "[nodeId=" + nodeId + ", msg=" + msg + ']'); |
| } |
| } |
| catch (InterruptedException ignored) { |
| Thread.currentThread().interrupt(); |
| |
| U.warn(log, "Failed to process fragmentizer communication message (thread was interrupted) "+ |
| "[nodeId=" + nodeId + ", msg=" + msg + ']'); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException { |
| while (!isCancelled()) { |
| IgniteBiTuple<UUID, IgfsCommunicationMessage> req = msgs.take(); |
| |
| UUID nodeId = req.get1(); |
| |
| if (req.get2() instanceof IgfsFragmentizerRequest) { |
| IgfsFragmentizerRequest fragmentizerReq = (IgfsFragmentizerRequest)req.get2(); |
| |
| if (!rw.tryReadLock()) { |
| if (log.isDebugEnabled()) |
| log.debug("Received fragmentizing request while stopping grid (will ignore) " + |
| "[nodeId=" + nodeId + ", req=" + req.get2() + ']'); |
| |
| continue; // while. |
| } |
| |
| try { |
| try { |
| processFragmentizerRequest(fragmentizerReq); |
| } |
| catch (IgniteCheckedException e) { |
| if (e.hasCause(ClusterTopologyCheckedException.class)) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to process fragmentizer request (remote node left the grid) " + |
| "[req=" + req + ", err=" + e.getMessage() + ']'); |
| } |
| else |
| U.error(log, "Failed to process fragmentizer request [nodeId=" + nodeId + |
| ", req=" + req + ']', e); |
| } |
| finally { |
| sendResponse(nodeId, new IgfsFragmentizerResponse(fragmentizerReq.fileId())); |
| } |
| } |
| finally { |
| rw.readUnlock(); |
| } |
| } |
| else { |
| assert req.get2() instanceof IgfsSyncMessage; |
| |
| IgfsSyncMessage syncMsg = (IgfsSyncMessage)req.get2(); |
| |
| if (!syncMsg.response()) { |
| IgfsSyncMessage res = new IgfsSyncMessage(syncMsg.order(), true); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Sending fragmentizer sync response to remote node [nodeId=" + nodeId + |
| ", res=" + res + ']'); |
| |
| sendResponse(nodeId, res); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Sends response to remote node. |
| * |
| * @param nodeId Node ID to send response to. |
| * @param msg Message to send. |
| */ |
| private void sendResponse(UUID nodeId, IgfsCommunicationMessage msg) { |
| try { |
| sendWithRetries(nodeId, msg); |
| } |
| catch (IgniteCheckedException e) { |
| if (e.hasCause(ClusterTopologyCheckedException.class)) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send sync response to IGFS fragmentizer coordinator " + |
| "(originating node left the grid): " + nodeId); |
| } |
| else |
| U.error(log, "Failed to send sync response to IGFS fragmentizer coordinator: " + nodeId, e); |
| } |
| } |
| } |
| |
| /** |
| * Hash set that overrides equals to use identity comparison. |
| */ |
| private static class IdentityHashSet extends GridConcurrentHashSet<UUID> { |
| /** */ |
| private static final long serialVersionUID = 0L; |
| |
| /** |
| * Constructor. |
| * |
| * @param c Collection to add. |
| */ |
| private IdentityHashSet(Collection<UUID> c) { |
| super(c); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean equals(Object o) { |
| // Identity comparison. |
| return this == o; |
| } |
| } |
| } |
