apache / carbondata / 816

Build # 816

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/367fe4cd78d358b04b2ab5a616027f1db5cf2a4d">367fe4cd7<a href="https://github.com/apache/carbondata/commit/367fe4cd78d358b04b2ab5a616027f1db5cf2a4d">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/aa37c4fa93ae4e1693119a1f528c095987629d73">aa37c4fa9</a>">aa37c4fa9</a><a href="https://github.com/apache/carbondata/commit/367fe4cd78d358b04b2ab5a616027f1db5cf2a4d"> into 8427771fc">8427771fc</a>

Run Details

83 of 83 new or added lines in 8 files covered. (100.0%)

60426 of 75044 relevant lines covered (80.52%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.48

/core/src/main/java/org/apache/carbondata/core/memory/UnsafeMemoryManager.java

/*
 * 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.carbondata.core.memory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;

/**
 * Manages memory for instance.
 */
public class UnsafeMemoryManager {

  private static final LogService LOGGER =
      LogServiceFactory.getLogService(UnsafeMemoryManager.class.getName());

  private static boolean offHeap = Boolean.parseBoolean(CarbonProperties.getInstance()
      .getProperty(CarbonCommonConstants.ENABLE_OFFHEAP_SORT,
          CarbonCommonConstants.ENABLE_OFFHEAP_SORT_DEFAULT));
  private static Map<Long,Set<MemoryBlock>> taskIdToMemoryBlockMap;
  static {
    long size = 0L;
    String defaultWorkingMemorySize = null;
    try {
      // check if driver unsafe memory is configured and JVM process is in driver. In that case
      // initialize unsafe memory configured for driver
      boolean isDriver = Boolean.parseBoolean(CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.IS_DRIVER_INSTANCE, "false"));
      boolean initializedWithUnsafeDriverMemory = false;
      if (isDriver) {
        defaultWorkingMemorySize = CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.UNSAFE_DRIVER_WORKING_MEMORY_IN_MB);
        if (null != defaultWorkingMemorySize) {
          size = Long.parseLong(defaultWorkingMemorySize);
          initializedWithUnsafeDriverMemory = true;
        }
      }
      if (!initializedWithUnsafeDriverMemory) {
        defaultWorkingMemorySize = CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);
        if (null != defaultWorkingMemorySize) {
          size = Long.parseLong(defaultWorkingMemorySize);
        }
      }
    } catch (Exception e) {
      LOGGER.info("Invalid memory size value: " + defaultWorkingMemorySize);
    }
    long takenSize = size;
    MemoryAllocator allocator;
    if (offHeap) {
      allocator = MemoryAllocator.UNSAFE;
      long defaultSize = Long.parseLong(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT);
      if (takenSize < defaultSize) {
        takenSize = defaultSize;
      }
      takenSize = takenSize * 1024 * 1024;
    } else {
      long maxMemory = Runtime.getRuntime().maxMemory() * 60 / 100;
      if (takenSize == 0L) {
        takenSize = maxMemory;
      } else {
        takenSize = takenSize * 1024 * 1024;
        if (takenSize > maxMemory) {
          takenSize = maxMemory;
        }
      }
      allocator = MemoryAllocator.HEAP;
    }
    INSTANCE = new UnsafeMemoryManager(takenSize, allocator);
    taskIdToMemoryBlockMap = new HashMap<>();
  }

  public static final UnsafeMemoryManager INSTANCE;

  private long totalMemory;

  private long memoryUsed;

  private MemoryAllocator allocator;

  private UnsafeMemoryManager(long totalMemory, MemoryAllocator allocator) {
    this.totalMemory = totalMemory;
    this.allocator = allocator;
    LOGGER
        .info("Working Memory manager is created with size " + totalMemory + " with " + allocator);
  }

  private synchronized MemoryBlock allocateMemory(MemoryAllocator memoryAllocator, long taskId,
      long memoryRequested) {
    if (memoryUsed + memoryRequested <= totalMemory) {
      MemoryBlock allocate = memoryAllocator.allocate(memoryRequested);
      memoryUsed += allocate.size();
      Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);
      if (null == listOfMemoryBlock) {
        listOfMemoryBlock = new HashSet<>();
        taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);
      }
      listOfMemoryBlock.add(allocate);
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Memory block (" + allocate + ") is created with size " + allocate.size()
            + ". Total memory used " + memoryUsed + "Bytes, left " + (totalMemory - memoryUsed)
            + "Bytes");
      }
      return allocate;
    }
    return null;
  }

  public synchronized void freeMemory(long taskId, MemoryBlock memoryBlock) {
    freeMemory(allocator, taskId, memoryBlock);
  }

  public synchronized void freeMemory(MemoryAllocator memoryAllocator, long taskId,
      MemoryBlock memoryBlock) {
    if (taskIdToMemoryBlockMap.containsKey(taskId)) {
      taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);
    }
    if (!memoryBlock.isFreedStatus()) {
      memoryAllocator.free(memoryBlock);
      memoryUsed -= memoryBlock.size();
      memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(
            "Freeing memory of size: " + memoryBlock.size() + "available memory:  " + (totalMemory
                - memoryUsed));
      }
    }
  }

  public synchronized void freeMemoryAll(long taskId) {
    Set<MemoryBlock> memoryBlockSet = null;
    memoryBlockSet = taskIdToMemoryBlockMap.remove(taskId);
    long occuppiedMemory = 0;
    if (null != memoryBlockSet) {
      Iterator<MemoryBlock> iterator = memoryBlockSet.iterator();
      MemoryBlock memoryBlock = null;
      while (iterator.hasNext()) {
        memoryBlock = iterator.next();
        if (!memoryBlock.isFreedStatus()) {
          occuppiedMemory += memoryBlock.size();
          allocator.free(memoryBlock);
        }
      }
    }
    memoryUsed -= occuppiedMemory;
    memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "Freeing memory of size: " + occuppiedMemory + ": Current available memory is: " + (
              totalMemory - memoryUsed));
    }
    LOGGER.info("Total memory used after task " + taskId + " is " + memoryUsed
        + " Current tasks running now are : " + taskIdToMemoryBlockMap.keySet());
  }

  public synchronized boolean isMemoryAvailable() {
    return memoryUsed > totalMemory;
  }

  public long getUsableMemory() {
    return totalMemory;
  }

  /**
   * It tries to allocate memory of `size` bytes, keep retry until it allocates successfully.
   */
  public static MemoryBlock allocateMemoryWithRetry(long taskId, long size)
      throws MemoryException {
    return allocateMemoryWithRetry(INSTANCE.allocator, taskId, size);
  }

  public static MemoryBlock allocateMemoryWithRetry(MemoryAllocator memoryAllocator, long taskId,
      long size) throws MemoryException {
    MemoryBlock baseBlock = null;
    int tries = 0;
    while (tries < 300) {
      baseBlock = INSTANCE.allocateMemory(memoryAllocator, taskId, size);
      if (baseBlock == null) {
        try {
          LOGGER.info("Memory is not available, retry after 500 millis");
          Thread.sleep(500);
        } catch (InterruptedException e) {
          throw new MemoryException(e);
        }
      } else {
        break;
      }
      tries++;
    }
    if (baseBlock == null) {
      INSTANCE.printCurrentMemoryUsage();
      throw new MemoryException(
          "Not enough memory. please increase carbon.unsafe.working.memory.in.mb");
    }
    return baseBlock;
  }

  public static boolean isOffHeap() {
    return offHeap;
  }

  private synchronized void printCurrentMemoryUsage() {
    LOGGER.error(
        " Memory Used : " + memoryUsed + " Tasks running : " + taskIdToMemoryBlockMap.keySet());
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.memory;
19
20	import java.util.HashMap;
21	import java.util.HashSet;
22	import java.util.Iterator;
23	import java.util.Map;
24	import java.util.Set;
25
26	import org.apache.carbondata.common.logging.LogService;
27	import org.apache.carbondata.common.logging.LogServiceFactory;
28	import org.apache.carbondata.core.constants.CarbonCommonConstants;
29	import org.apache.carbondata.core.util.CarbonProperties;
30
31	/**
32	* Manages memory for instance.
33	*/
34	public class UnsafeMemoryManager {
35
36	private static final LogService LOGGER =	2✔
37	LogServiceFactory.getLogService(UnsafeMemoryManager.class.getName());	2✔
38
39	private static boolean offHeap = Boolean.parseBoolean(CarbonProperties.getInstance()	2✔
40	.getProperty(CarbonCommonConstants.ENABLE_OFFHEAP_SORT,	2✔
41	CarbonCommonConstants.ENABLE_OFFHEAP_SORT_DEFAULT));
42	private static Map<Long,Set<MemoryBlock>> taskIdToMemoryBlockMap;
43	static {
44	long size = 0L;	2✔
45	String defaultWorkingMemorySize = null;	2✔
46	try {
47	// check if driver unsafe memory is configured and JVM process is in driver. In that case
48	// initialize unsafe memory configured for driver
49	boolean isDriver = Boolean.parseBoolean(CarbonProperties.getInstance()	2✔
50	.getProperty(CarbonCommonConstants.IS_DRIVER_INSTANCE, "false"));	2✔
51	boolean initializedWithUnsafeDriverMemory = false;	2✔
52	if (isDriver) {	2✔
53	defaultWorkingMemorySize = CarbonProperties.getInstance()	1✔
54	.getProperty(CarbonCommonConstants.UNSAFE_DRIVER_WORKING_MEMORY_IN_MB);	1✔
55	if (null != defaultWorkingMemorySize) {	1✔
56	size = Long.parseLong(defaultWorkingMemorySize);	×
57	initializedWithUnsafeDriverMemory = true;	×
58	}
59	}
60	if (!initializedWithUnsafeDriverMemory) {	2✔
61	defaultWorkingMemorySize = CarbonProperties.getInstance()	2✔
62	.getProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);	2✔
63	if (null != defaultWorkingMemorySize) {	2✔
64	size = Long.parseLong(defaultWorkingMemorySize);	×
65	}
66	}
67	} catch (Exception e) {	×
68	LOGGER.info("Invalid memory size value: " + defaultWorkingMemorySize);	×
69	}	2✔
70	long takenSize = size;	2✔
71	MemoryAllocator allocator;
72	if (offHeap) {	2✔
73	allocator = MemoryAllocator.UNSAFE;	2✔
74	long defaultSize = Long.parseLong(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT);	2✔
75	if (takenSize < defaultSize) {	2✔
76	takenSize = defaultSize;	2✔
77	}
78	takenSize = takenSize * 1024 * 1024;	2✔
79	} else {	2✔
80	long maxMemory = Runtime.getRuntime().maxMemory() * 60 / 100;	×
81	if (takenSize == 0L) {	×
82	takenSize = maxMemory;	×
83	} else {
84	takenSize = takenSize * 1024 * 1024;	×
85	if (takenSize > maxMemory) {	×
86	takenSize = maxMemory;	×
87	}
88	}
89	allocator = MemoryAllocator.HEAP;	×
90	}
91	INSTANCE = new UnsafeMemoryManager(takenSize, allocator);	2✔
92	taskIdToMemoryBlockMap = new HashMap<>();	2✔
93	}	2✔
94
95	public static final UnsafeMemoryManager INSTANCE;
96
97	private long totalMemory;
98
99	private long memoryUsed;
100
101	private MemoryAllocator allocator;
102
103	private UnsafeMemoryManager(long totalMemory, MemoryAllocator allocator) {	2✔
104	this.totalMemory = totalMemory;	2✔
105	this.allocator = allocator;	2✔
106	LOGGER	2✔
107	.info("Working Memory manager is created with size " + totalMemory + " with " + allocator);	2✔
108	}	2✔
109
110	private synchronized MemoryBlock allocateMemory(MemoryAllocator memoryAllocator, long taskId,
111	long memoryRequested) {
112	if (memoryUsed + memoryRequested <= totalMemory) {	2✔
113	MemoryBlock allocate = memoryAllocator.allocate(memoryRequested);	2✔
114	memoryUsed += allocate.size();	2✔
115	Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);	2✔
116	if (null == listOfMemoryBlock) {	2✔
117	listOfMemoryBlock = new HashSet<>();	2✔
118	taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);	2✔
119	}
120	listOfMemoryBlock.add(allocate);	2✔
121	if (LOGGER.isDebugEnabled()) {	2✔
122	LOGGER.debug("Memory block (" + allocate + ") is created with size " + allocate.size()	2✔
123	+ ". Total memory used " + memoryUsed + "Bytes, left " + (totalMemory - memoryUsed)
124	+ "Bytes");
125	}
126	return allocate;	2✔
127	}
128	return null;	×
129	}
130
131	public synchronized void freeMemory(long taskId, MemoryBlock memoryBlock) {
132	freeMemory(allocator, taskId, memoryBlock);	2✔
133	}	2✔
134
135	public synchronized void freeMemory(MemoryAllocator memoryAllocator, long taskId,
136	MemoryBlock memoryBlock) {
137	if (taskIdToMemoryBlockMap.containsKey(taskId)) {	2✔
138	taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);	2✔
139	}
140	if (!memoryBlock.isFreedStatus()) {	2✔
141	memoryAllocator.free(memoryBlock);	2✔
142	memoryUsed -= memoryBlock.size();	2✔
143	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	2✔
144	if (LOGGER.isDebugEnabled()) {	2✔
145	LOGGER.debug(	2✔
146	"Freeing memory of size: " + memoryBlock.size() + "available memory: " + (totalMemory	2✔
147	- memoryUsed));
148	}
149	}
150	}	2✔
151
152	public synchronized void freeMemoryAll(long taskId) {
153	Set<MemoryBlock> memoryBlockSet = null;	1✔
154	memoryBlockSet = taskIdToMemoryBlockMap.remove(taskId);	1✔
155	long occuppiedMemory = 0;	1✔
156	if (null != memoryBlockSet) {	1✔
157	Iterator<MemoryBlock> iterator = memoryBlockSet.iterator();	1✔
158	MemoryBlock memoryBlock = null;	1✔
159	while (iterator.hasNext()) {	1✔
160	memoryBlock = iterator.next();	1✔
161	if (!memoryBlock.isFreedStatus()) {	1✔
162	occuppiedMemory += memoryBlock.size();	1✔
163	allocator.free(memoryBlock);	1✔
164	}
165	}
166	}
167	memoryUsed -= occuppiedMemory;	1✔
168	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	1✔
169	if (LOGGER.isDebugEnabled()) {	1✔
170	LOGGER.debug(	1✔
171	"Freeing memory of size: " + occuppiedMemory + ": Current available memory is: " + (
172	totalMemory - memoryUsed));
173	}
174	LOGGER.info("Total memory used after task " + taskId + " is " + memoryUsed	1✔
175	+ " Current tasks running now are : " + taskIdToMemoryBlockMap.keySet());	1✔
176	}	1✔
177
178	public synchronized boolean isMemoryAvailable() {
179	return memoryUsed > totalMemory;	×
180	}
181
182	public long getUsableMemory() {
183	return totalMemory;	1✔
184	}
185
186	/**
187	* It tries to allocate memory of `size` bytes, keep retry until it allocates successfully.
188	*/
189	public static MemoryBlock allocateMemoryWithRetry(long taskId, long size)
190	throws MemoryException {
191	return allocateMemoryWithRetry(INSTANCE.allocator, taskId, size);	2✔
192	}
193
194	public static MemoryBlock allocateMemoryWithRetry(MemoryAllocator memoryAllocator, long taskId,
195	long size) throws MemoryException {
196	MemoryBlock baseBlock = null;	2✔
197	int tries = 0;	2✔
198	while (tries < 300) {	2✔
199	baseBlock = INSTANCE.allocateMemory(memoryAllocator, taskId, size);	2✔
200	if (baseBlock == null) {	2✔
201	try {
202	LOGGER.info("Memory is not available, retry after 500 millis");	×
203	Thread.sleep(500);	×
204	} catch (InterruptedException e) {	×
205	throw new MemoryException(e);	×
206	}	×
207	} else {
208	break;
209	}
210	tries++;	×
211	}
212	if (baseBlock == null) {	2✔
213	INSTANCE.printCurrentMemoryUsage();	×
214	throw new MemoryException(	×
215	"Not enough memory. please increase carbon.unsafe.working.memory.in.mb");
216	}
217	return baseBlock;	2✔
218	}
219
220	public static boolean isOffHeap() {
221	return offHeap;	1✔
222	}
223
224	private synchronized void printCurrentMemoryUsage() {
225	LOGGER.error(	×
226	" Memory Used : " + memoryUsed + " Tasks running : " + taskIdToMemoryBlockMap.keySet());	×
227	}	×
228	}

apache / carbondata / 816

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous