geode-core/src/main/java/org/apache/geode/internal/cache/OverflowOplogSet.java - geode - Git at Google

 /*
  * 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.geode.internal.cache;

 import java.io.IOException;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.cache.DiskAccessException;
 import org.apache.geode.internal.cache.entries.DiskEntry;
 import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
 import org.apache.geode.internal.logging.LogService;

 public class OverflowOplogSet implements OplogSet {
   private static final Logger logger = LogService.getLogger();

   private final AtomicInteger overflowOplogId = new AtomicInteger(0);
   private OverflowOplog lastOverflowWrite;
   private final ConcurrentMap<Integer, OverflowOplog> overflowMap = new ConcurrentHashMap<>();
   private final Map<Integer, OverflowOplog> compactibleOverflowMap = new LinkedHashMap<>();

   private int lastOverflowDir = 0;

   private DiskStoreImpl parent;

   public OverflowOplogSet(DiskStoreImpl parent) {
     this.parent = parent;
   }

   OverflowOplog getActiveOverflowOplog() {
     return this.lastOverflowWrite;
   }

   @Override
   public void modify(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
     DiskRegion dr = region.getDiskRegion();
     synchronized (this.overflowMap) {
       if (this.lastOverflowWrite != null) {
         if (this.lastOverflowWrite.modify(dr, entry, value, async)) {
           return;
         }
       }
       // Create a new one and put it on the front of the list.
       OverflowOplog oo = createOverflowOplog(value.getLength());
       addOverflow(oo);
       this.lastOverflowWrite = oo;
       boolean didIt = oo.modify(dr, entry, value, async);
       assert didIt;
     }
   }

   private long getMaxOplogSizeInBytes() {
     return parent.getMaxOplogSizeInBytes();
   }

   private DirectoryHolder[] getDirectories() {
     return parent.directories;
   }

   /**
    * @param minSize the minimum size this oplog can be
    */
   private OverflowOplog createOverflowOplog(long minSize) {
     lastOverflowDir++;
     if (lastOverflowDir >= getDirectories().length) {
       lastOverflowDir = 0;
     }
     int idx = -1;
     long maxOplogSizeParam = getMaxOplogSizeInBytes();
     if (maxOplogSizeParam < minSize) {
       maxOplogSizeParam = minSize;
     }

     // first look for a directory that has room for maxOplogSize
     for (int i = lastOverflowDir; i < getDirectories().length; i++) {
       long availableSpace = getDirectories()[i].getAvailableSpace();
       if (availableSpace >= maxOplogSizeParam) {
         idx = i;
         break;
       }
     }
     if (idx == -1 && lastOverflowDir != 0) {
       for (int i = 0; i < lastOverflowDir; i++) {
         long availableSpace = getDirectories()[i].getAvailableSpace();
         if (availableSpace >= maxOplogSizeParam) {
           idx = i;
           break;
         }
       }
     }

     if (idx == -1) {
       // if we couldn't find one big enough for the max look for one
       // that has min room
       for (int i = lastOverflowDir; i < getDirectories().length; i++) {
         long availableSpace = getDirectories()[i].getAvailableSpace();
         if (availableSpace >= minSize) {
           idx = i;
           break;
         }
       }
       if (idx == -1 && lastOverflowDir != 0) {
         for (int i = 0; i < lastOverflowDir; i++) {
           long availableSpace = getDirectories()[i].getAvailableSpace();
           if (availableSpace >= minSize) {
             idx = i;
             break;
           }
         }
       }
     }

     if (idx == -1) {
       if (parent.isCompactionEnabled()) { // fix for bug 41835
         idx = lastOverflowDir;
         if (getDirectories()[idx].getAvailableSpace() < minSize) {
           logger.warn(
               "Even though the configured directory size limit has been exceeded a new oplog will be created because compaction is enabled. The configured limit is {}. The current space used in the directory by this disk store is {}.",
               new Object[] {Long.valueOf(getDirectories()[idx].getUsedSpace()),
                   Long.valueOf(getDirectories()[idx].getCapacity())});
         }
       } else {
         throw new DiskAccessException(
             String.format(
                 "Directories are full, not able to accommodate this operation.Switching problem for entry having DiskID= %s",
                 "needed " + minSize + " bytes"),
             parent);
       }
     }
     int id = this.overflowOplogId.incrementAndGet();
     lastOverflowDir = idx;
     return new OverflowOplog(id, this, getDirectories()[idx], minSize);
   }

   void addOverflow(OverflowOplog oo) {
     this.overflowMap.put(oo.getOplogId(), oo);
   }

   void removeOverflow(OverflowOplog oo) {
     if (!basicRemoveOverflow(oo)) {
       synchronized (this.compactibleOverflowMap) {
         this.compactibleOverflowMap.remove(oo.getOplogId());
       }
     }
   }

   boolean basicRemoveOverflow(OverflowOplog oo) {
     if (this.lastOverflowWrite == oo) {
       this.lastOverflowWrite = null;
     }
     return this.overflowMap.remove(oo.getOplogId(), oo);
   }

   public void closeOverflow() {
     for (OverflowOplog oo : this.overflowMap.values()) {
       oo.destroy();
     }
     synchronized (this.compactibleOverflowMap) {
       for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
         oo.destroy();
       }
     }
   }

   private void removeOverflow(DiskRegion dr, DiskEntry entry) {
     // find the overflow oplog that it is currently in and remove the entry from it
     DiskId id = entry.getDiskId();
     synchronized (id) {
       long oplogId = id.setOplogId(-1);
       if (oplogId != -1) {
         synchronized (this.overflowMap) { // to prevent concurrent remove see bug 41646
           OverflowOplog oplog = getChild((int) oplogId);
           if (oplog != null) {
             oplog.remove(dr, entry);
           }
         }
       }
     }
   }

   void copyForwardForOverflowCompact(DiskEntry de, byte[] valueBytes, int length, byte userBits) {
     synchronized (this.overflowMap) {
       if (this.lastOverflowWrite != null) {
         if (this.lastOverflowWrite.copyForwardForOverflowCompact(de, valueBytes, length,
             userBits)) {
           return;
         }
       }
       OverflowOplog oo = createOverflowOplog(length);
       this.lastOverflowWrite = oo;
       addOverflow(oo);
       boolean didIt = oo.copyForwardForOverflowCompact(de, valueBytes, length, userBits);
       assert didIt;
     }
   }

   @Override
   public OverflowOplog getChild(long oplogId) {
     // the oplog id is cast to an integer because the overflow
     // map uses integer oplog ids.
     return getChild((int) oplogId);
   }

   public OverflowOplog getChild(int oplogId) {
     OverflowOplog result = this.overflowMap.get(oplogId);
     if (result == null) {
       synchronized (this.compactibleOverflowMap) {
         result = this.compactibleOverflowMap.get(oplogId);
       }
     }
     return result;
   }

   @Override
   public void create(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
     modify(region, entry, value, async);
   }

   @Override
   public void remove(InternalRegion region, DiskEntry entry, boolean async, boolean isClear) {
     removeOverflow(region.getDiskRegion(), entry);
   }

   void addOverflowToBeCompacted(OverflowOplog oplog) {
     synchronized (this.compactibleOverflowMap) {
       this.compactibleOverflowMap.put(oplog.getOplogId(), oplog);
     }
     basicRemoveOverflow(oplog);
     parent.scheduleCompaction();
   }

   public void getCompactableOplogs(List<CompactableOplog> l, int max) {
     synchronized (this.compactibleOverflowMap) {
       Iterator<OverflowOplog> itr = this.compactibleOverflowMap.values().iterator();
       while (itr.hasNext() && l.size() < max) {
         OverflowOplog oplog = itr.next();
         if (oplog.needsCompaction()) {
           l.add(oplog);
         }
       }
     }
   }

   void testHookCloseAllOverflowChannels() {
     synchronized (this.overflowMap) {
       for (OverflowOplog oo : this.overflowMap.values()) {
         FileChannel oplogFileChannel = oo.getFileChannel();
         try {
           oplogFileChannel.close();
         } catch (IOException ignore) {
         }
       }
     }
     synchronized (this.compactibleOverflowMap) {
       for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
         FileChannel oplogFileChannel = oo.getFileChannel();
         try {
           oplogFileChannel.close();
         } catch (IOException ignore) {
         }
       }
     }
   }

   ArrayList<OverflowOplog> testHookGetAllOverflowOplogs() {
     ArrayList<OverflowOplog> result = new ArrayList<>();
     synchronized (this.overflowMap) {
       for (OverflowOplog oo : this.overflowMap.values()) {
         result.add(oo);
       }
     }
     synchronized (this.compactibleOverflowMap) {
       for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
         result.add(oo);
       }
     }

     return result;
   }

   void testHookCloseAllOverflowOplogs() {
     synchronized (this.overflowMap) {
       for (OverflowOplog oo : this.overflowMap.values()) {
         oo.close();
       }
     }
     synchronized (this.compactibleOverflowMap) {
       for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
         oo.close();
       }
     }
   }

   public DiskStoreImpl getParent() {
     return parent;
   }
 }
	/*
	* 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.geode.internal.cache;

	import java.io.IOException;
	import java.nio.channels.FileChannel;
	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.LinkedHashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;
	import java.util.concurrent.atomic.AtomicInteger;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.cache.DiskAccessException;
	import org.apache.geode.internal.cache.entries.DiskEntry;
	import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
	import org.apache.geode.internal.logging.LogService;

	public class OverflowOplogSet implements OplogSet {
	private static final Logger logger = LogService.getLogger();

	private final AtomicInteger overflowOplogId = new AtomicInteger(0);
	private OverflowOplog lastOverflowWrite;
	private final ConcurrentMap<Integer, OverflowOplog> overflowMap = new ConcurrentHashMap<>();
	private final Map<Integer, OverflowOplog> compactibleOverflowMap = new LinkedHashMap<>();

	private int lastOverflowDir = 0;

	private DiskStoreImpl parent;

	public OverflowOplogSet(DiskStoreImpl parent) {
	this.parent = parent;
	}

	OverflowOplog getActiveOverflowOplog() {
	return this.lastOverflowWrite;
	}

	@Override
	public void modify(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
	DiskRegion dr = region.getDiskRegion();
	synchronized (this.overflowMap) {
	if (this.lastOverflowWrite != null) {
	if (this.lastOverflowWrite.modify(dr, entry, value, async)) {
	return;
	}
	}
	// Create a new one and put it on the front of the list.
	OverflowOplog oo = createOverflowOplog(value.getLength());
	addOverflow(oo);
	this.lastOverflowWrite = oo;
	boolean didIt = oo.modify(dr, entry, value, async);
	assert didIt;
	}
	}

	private long getMaxOplogSizeInBytes() {
	return parent.getMaxOplogSizeInBytes();
	}

	private DirectoryHolder[] getDirectories() {
	return parent.directories;
	}

	/**
	* @param minSize the minimum size this oplog can be
	*/
	private OverflowOplog createOverflowOplog(long minSize) {
	lastOverflowDir++;
	if (lastOverflowDir >= getDirectories().length) {
	lastOverflowDir = 0;
	}
	int idx = -1;
	long maxOplogSizeParam = getMaxOplogSizeInBytes();
	if (maxOplogSizeParam < minSize) {
	maxOplogSizeParam = minSize;
	}

	// first look for a directory that has room for maxOplogSize
	for (int i = lastOverflowDir; i < getDirectories().length; i++) {
	long availableSpace = getDirectories()[i].getAvailableSpace();
	if (availableSpace >= maxOplogSizeParam) {
	idx = i;
	break;
	}
	}
	if (idx == -1 && lastOverflowDir != 0) {
	for (int i = 0; i < lastOverflowDir; i++) {
	long availableSpace = getDirectories()[i].getAvailableSpace();
	if (availableSpace >= maxOplogSizeParam) {
	idx = i;
	break;
	}
	}
	}

	if (idx == -1) {
	// if we couldn't find one big enough for the max look for one
	// that has min room
	for (int i = lastOverflowDir; i < getDirectories().length; i++) {
	long availableSpace = getDirectories()[i].getAvailableSpace();
	if (availableSpace >= minSize) {
	idx = i;
	break;
	}
	}
	if (idx == -1 && lastOverflowDir != 0) {
	for (int i = 0; i < lastOverflowDir; i++) {
	long availableSpace = getDirectories()[i].getAvailableSpace();
	if (availableSpace >= minSize) {
	idx = i;
	break;
	}
	}
	}
	}

	if (idx == -1) {
	if (parent.isCompactionEnabled()) { // fix for bug 41835
	idx = lastOverflowDir;
	if (getDirectories()[idx].getAvailableSpace() < minSize) {
	logger.warn(
	"Even though the configured directory size limit has been exceeded a new oplog will be created because compaction is enabled. The configured limit is {}. The current space used in the directory by this disk store is {}.",
	new Object[] {Long.valueOf(getDirectories()[idx].getUsedSpace()),
	Long.valueOf(getDirectories()[idx].getCapacity())});
	}
	} else {
	throw new DiskAccessException(
	String.format(
	"Directories are full, not able to accommodate this operation.Switching problem for entry having DiskID= %s",
	"needed " + minSize + " bytes"),
	parent);
	}
	}
	int id = this.overflowOplogId.incrementAndGet();
	lastOverflowDir = idx;
	return new OverflowOplog(id, this, getDirectories()[idx], minSize);
	}

	void addOverflow(OverflowOplog oo) {
	this.overflowMap.put(oo.getOplogId(), oo);
	}

	void removeOverflow(OverflowOplog oo) {
	if (!basicRemoveOverflow(oo)) {
	synchronized (this.compactibleOverflowMap) {
	this.compactibleOverflowMap.remove(oo.getOplogId());
	}
	}
	}

	boolean basicRemoveOverflow(OverflowOplog oo) {
	if (this.lastOverflowWrite == oo) {
	this.lastOverflowWrite = null;
	}
	return this.overflowMap.remove(oo.getOplogId(), oo);
	}

	public void closeOverflow() {
	for (OverflowOplog oo : this.overflowMap.values()) {
	oo.destroy();
	}
	synchronized (this.compactibleOverflowMap) {
	for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
	oo.destroy();
	}
	}
	}

	private void removeOverflow(DiskRegion dr, DiskEntry entry) {
	// find the overflow oplog that it is currently in and remove the entry from it
	DiskId id = entry.getDiskId();
	synchronized (id) {
	long oplogId = id.setOplogId(-1);
	if (oplogId != -1) {
	synchronized (this.overflowMap) { // to prevent concurrent remove see bug 41646
	OverflowOplog oplog = getChild((int) oplogId);
	if (oplog != null) {
	oplog.remove(dr, entry);
	}
	}
	}
	}
	}

	void copyForwardForOverflowCompact(DiskEntry de, byte[] valueBytes, int length, byte userBits) {
	synchronized (this.overflowMap) {
	if (this.lastOverflowWrite != null) {
	if (this.lastOverflowWrite.copyForwardForOverflowCompact(de, valueBytes, length,
	userBits)) {
	return;
	}
	}
	OverflowOplog oo = createOverflowOplog(length);
	this.lastOverflowWrite = oo;
	addOverflow(oo);
	boolean didIt = oo.copyForwardForOverflowCompact(de, valueBytes, length, userBits);
	assert didIt;
	}
	}

	@Override
	public OverflowOplog getChild(long oplogId) {
	// the oplog id is cast to an integer because the overflow
	// map uses integer oplog ids.
	return getChild((int) oplogId);
	}

	public OverflowOplog getChild(int oplogId) {
	OverflowOplog result = this.overflowMap.get(oplogId);
	if (result == null) {
	synchronized (this.compactibleOverflowMap) {
	result = this.compactibleOverflowMap.get(oplogId);
	}
	}
	return result;
	}

	@Override
	public void create(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
	modify(region, entry, value, async);
	}

	@Override
	public void remove(InternalRegion region, DiskEntry entry, boolean async, boolean isClear) {
	removeOverflow(region.getDiskRegion(), entry);
	}

	void addOverflowToBeCompacted(OverflowOplog oplog) {
	synchronized (this.compactibleOverflowMap) {
	this.compactibleOverflowMap.put(oplog.getOplogId(), oplog);
	}
	basicRemoveOverflow(oplog);
	parent.scheduleCompaction();
	}

	public void getCompactableOplogs(List<CompactableOplog> l, int max) {
	synchronized (this.compactibleOverflowMap) {
	Iterator<OverflowOplog> itr = this.compactibleOverflowMap.values().iterator();
	while (itr.hasNext() && l.size() < max) {
	OverflowOplog oplog = itr.next();
	if (oplog.needsCompaction()) {
	l.add(oplog);
	}
	}
	}
	}

	void testHookCloseAllOverflowChannels() {
	synchronized (this.overflowMap) {
	for (OverflowOplog oo : this.overflowMap.values()) {
	FileChannel oplogFileChannel = oo.getFileChannel();
	try {
	oplogFileChannel.close();
	} catch (IOException ignore) {
	}
	}
	}
	synchronized (this.compactibleOverflowMap) {
	for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
	FileChannel oplogFileChannel = oo.getFileChannel();
	try {
	oplogFileChannel.close();
	} catch (IOException ignore) {
	}
	}
	}
	}

	ArrayList<OverflowOplog> testHookGetAllOverflowOplogs() {
	ArrayList<OverflowOplog> result = new ArrayList<>();
	synchronized (this.overflowMap) {
	for (OverflowOplog oo : this.overflowMap.values()) {
	result.add(oo);
	}
	}
	synchronized (this.compactibleOverflowMap) {
	for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
	result.add(oo);
	}
	}

	return result;
	}

	void testHookCloseAllOverflowOplogs() {
	synchronized (this.overflowMap) {
	for (OverflowOplog oo : this.overflowMap.values()) {
	oo.close();
	}
	}
	synchronized (this.compactibleOverflowMap) {
	for (OverflowOplog oo : this.compactibleOverflowMap.values()) {
	oo.close();
	}
	}
	}

	public DiskStoreImpl getParent() {
	return parent;
	}
	}