1232

Build # 1232

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/6561cc9fae7b6c17c44adf2c62b176f3f009b65e">6561cc9fa<a href="https://github.com/apache/carbondata/commit/6561cc9fae7b6c17c44adf2c62b176f3f009b65e">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/14517a44932d41eedb515590c6d290d5047d5aac">14517a449</a>">14517a449</a><a href="https://github.com/apache/carbondata/commit/6561cc9fae7b6c17c44adf2c62b176f3f009b65e"> into 9578786b2">9578786b2</a>

Run Details

12 of 12 new or added lines in 5 files covered. (100.0%)

60529 of 75118 relevant lines covered (80.58%)

1.07 hits per line

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

78.26

/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.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

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

  private static final Logger 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<String,Set<MemoryBlock>> taskIdToMemoryBlockMap;
  static {
    long size = 0L;
    String configuredWorkingMemorySize = 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) {
        configuredWorkingMemorySize = CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.UNSAFE_DRIVER_WORKING_MEMORY_IN_MB);
        if (null != configuredWorkingMemorySize) {
          size = Long.parseLong(configuredWorkingMemorySize);
          initializedWithUnsafeDriverMemory = true;
        }
      }
      if (!initializedWithUnsafeDriverMemory) {
        configuredWorkingMemorySize = CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);
        if (null != configuredWorkingMemorySize) {
          size = Long.parseLong(configuredWorkingMemorySize);
        }
      }
    } catch (Exception e) {
      LOGGER.info("Invalid working memory size value: " + configuredWorkingMemorySize);
    }
    long takenSize = size;
    MemoryType memoryType;
    if (offHeap) {
      memoryType = MemoryType.OFFHEAP;
      long defaultSize = Long.parseLong(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT);
      if (takenSize < defaultSize) {
        takenSize = defaultSize;
        LOGGER.warn(String.format(
            "It is not recommended to set unsafe working memory size less than %sMB,"
                + " so setting default value to %d",
            CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT, 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;
        }
      }
      memoryType = MemoryType.ONHEAP;
    }
    INSTANCE = new UnsafeMemoryManager(takenSize, memoryType);
    taskIdToMemoryBlockMap = new HashMap<>();
  }

  public static final UnsafeMemoryManager INSTANCE;

  private long totalMemory;

  private long memoryUsed;

  private MemoryType memoryType;

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

  private synchronized MemoryBlock allocateMemory(MemoryType memoryType, String taskId,
      long memoryRequested) {
    if (memoryUsed + memoryRequested <= totalMemory) {
      MemoryBlock memoryBlock = getMemoryAllocator(memoryType).allocate(memoryRequested);
      memoryUsed += memoryBlock.size();
      Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);
      if (null == listOfMemoryBlock) {
        listOfMemoryBlock = new HashSet<>();
        taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);
      }
      listOfMemoryBlock.add(memoryBlock);
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(String.format("Creating working Memory block (%s) with size %d."
                + " Total memory used %d Bytes, left %d Bytes.",
            memoryBlock.toString(), memoryBlock.size(), memoryUsed, totalMemory - memoryUsed));
      }
      return memoryBlock;
    }
    return null;
  }

  public synchronized void freeMemory(String taskId, MemoryBlock memoryBlock) {
    if (taskIdToMemoryBlockMap.containsKey(taskId)) {
      taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);
    }
    if (!memoryBlock.isFreedStatus()) {
      getMemoryAllocator(memoryBlock.getMemoryType()).free(memoryBlock);
      memoryUsed -= memoryBlock.size();
      memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(String.format(
            "Freeing working memory block (%s) with size: %d, current available memory is: %d",
            memoryBlock.toString(), memoryBlock.size(), totalMemory - memoryUsed));
      }
    }
  }

  public synchronized void freeMemoryAll(String 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();
          getMemoryAllocator(memoryBlock.getMemoryType()).free(memoryBlock);
        }
      }
    }
    memoryUsed -= occuppiedMemory;
    memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(String.format(
          "Freeing working memory of size %d. Current available memory is %d",
          occuppiedMemory, totalMemory - memoryUsed));
    }
    LOGGER.info(String.format(
        "Total working memory used after task %s is %d. Current running tasks are %s",
        taskId, memoryUsed, StringUtils.join(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(String taskId, long size)
      throws MemoryException {
    return allocateMemoryWithRetry(INSTANCE.memoryType, taskId, size);
  }

  public static MemoryBlock allocateMemoryWithRetry(MemoryType memoryType, String taskId,
      long size) throws MemoryException {
    MemoryBlock baseBlock = null;
    int tries = 0;
    while (tries < 300) {
      baseBlock = INSTANCE.allocateMemory(memoryType, taskId, size);
      if (baseBlock == null) {
        try {
          LOGGER.info("Working memory is not available, retry after 500 ms");
          Thread.sleep(500);
        } catch (InterruptedException e) {
          throw new MemoryException(e);
        }
      } else {
        break;
      }
      tries++;
    }
    if (baseBlock == null) {
      INSTANCE.printCurrentMemoryUsage();
      throw new MemoryException("Not enough working memory, please increase "
          + CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);
    }
    return baseBlock;
  }

  private MemoryAllocator getMemoryAllocator(MemoryType memoryType) {
    switch (memoryType) {
      case ONHEAP:
        return MemoryAllocator.HEAP;
      default:
        return MemoryAllocator.UNSAFE;
    }
  }

  public static boolean isOffHeap() {
    return offHeap;
  }

  private synchronized void printCurrentMemoryUsage() {
    LOGGER.info(String.format("Working memory used %d, running tasks are %s",
        memoryUsed, StringUtils.join(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.LogServiceFactory;
27	import org.apache.carbondata.core.constants.CarbonCommonConstants;
28	import org.apache.carbondata.core.util.CarbonProperties;
29
30	import org.apache.commons.lang3.StringUtils;
31	import org.apache.log4j.Logger;
32
33	/**
34	* Manages memory for instance.
35	*/
36	public class UnsafeMemoryManager {
37
38	private static final Logger LOGGER =	2✔
39	LogServiceFactory.getLogService(UnsafeMemoryManager.class.getName());	2✔
40
41	private static boolean offHeap = Boolean.parseBoolean(CarbonProperties.getInstance()	2✔
42	.getProperty(CarbonCommonConstants.ENABLE_OFFHEAP_SORT,	2✔
43	CarbonCommonConstants.ENABLE_OFFHEAP_SORT_DEFAULT));
44	private static Map<String,Set<MemoryBlock>> taskIdToMemoryBlockMap;
45	static {
46	long size = 0L;	2✔
47	String configuredWorkingMemorySize = null;	2✔
48	try {
49	// check if driver unsafe memory is configured and JVM process is in driver. In that case
50	// initialize unsafe memory configured for driver
51	boolean isDriver = Boolean.parseBoolean(CarbonProperties.getInstance()	2✔
52	.getProperty(CarbonCommonConstants.IS_DRIVER_INSTANCE, "false"));	2✔
53	boolean initializedWithUnsafeDriverMemory = false;	2✔
54	if (isDriver) {	2✔
55	configuredWorkingMemorySize = CarbonProperties.getInstance()	1✔
56	.getProperty(CarbonCommonConstants.UNSAFE_DRIVER_WORKING_MEMORY_IN_MB);	1✔
57	if (null != configuredWorkingMemorySize) {	1✔
58	size = Long.parseLong(configuredWorkingMemorySize);	×
59	initializedWithUnsafeDriverMemory = true;	×
60	}
61	}
62	if (!initializedWithUnsafeDriverMemory) {	2✔
63	configuredWorkingMemorySize = CarbonProperties.getInstance()	2✔
64	.getProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);	2✔
65	if (null != configuredWorkingMemorySize) {	2✔
66	size = Long.parseLong(configuredWorkingMemorySize);	×
67	}
68	}
69	} catch (Exception e) {	×
70	LOGGER.info("Invalid working memory size value: " + configuredWorkingMemorySize);	×
71	}	2✔
72	long takenSize = size;	2✔
73	MemoryType memoryType;
74	if (offHeap) {	2✔
75	memoryType = MemoryType.OFFHEAP;	2✔
76	long defaultSize = Long.parseLong(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT);	2✔
77	if (takenSize < defaultSize) {	2✔
78	takenSize = defaultSize;	2✔
79	LOGGER.warn(String.format(	2✔
80	"It is not recommended to set unsafe working memory size less than %sMB,"
81	+ " so setting default value to %d",
82	CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB_DEFAULT, defaultSize));	2✔
83	}
84	takenSize = takenSize * 1024 * 1024;	2✔
85	} else {	2✔
86	long maxMemory = Runtime.getRuntime().maxMemory() * 60 / 100;	×
87	if (takenSize == 0L) {	×
88	takenSize = maxMemory;	×
89	} else {
90	takenSize = takenSize * 1024 * 1024;	×
91	if (takenSize > maxMemory) {	×
92	takenSize = maxMemory;	×
93	}
94	}
95	memoryType = MemoryType.ONHEAP;	×
96	}
97	INSTANCE = new UnsafeMemoryManager(takenSize, memoryType);	2✔
98	taskIdToMemoryBlockMap = new HashMap<>();	2✔
99	}	2✔
100
101	public static final UnsafeMemoryManager INSTANCE;
102
103	private long totalMemory;
104
105	private long memoryUsed;
106
107	private MemoryType memoryType;
108
109	private UnsafeMemoryManager(long totalMemory, MemoryType memoryType) {	2✔
110	this.totalMemory = totalMemory;	2✔
111	this.memoryType = memoryType;	2✔
112	LOGGER.info(	2✔
113	"Working Memory manager is created with size " + totalMemory + " with " + memoryType);
114	}	2✔
115
116	private synchronized MemoryBlock allocateMemory(MemoryType memoryType, String taskId,
117	long memoryRequested) {
118	if (memoryUsed + memoryRequested <= totalMemory) {	2✔
119	MemoryBlock memoryBlock = getMemoryAllocator(memoryType).allocate(memoryRequested);	2✔
120	memoryUsed += memoryBlock.size();	2✔
121	Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);	2✔
122	if (null == listOfMemoryBlock) {	2✔
123	listOfMemoryBlock = new HashSet<>();	2✔
124	taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);	2✔
125	}
126	listOfMemoryBlock.add(memoryBlock);	2✔
127	if (LOGGER.isDebugEnabled()) {	2✔
128	LOGGER.debug(String.format("Creating working Memory block (%s) with size %d."	2✔
129	+ " Total memory used %d Bytes, left %d Bytes.",
130	memoryBlock.toString(), memoryBlock.size(), memoryUsed, totalMemory - memoryUsed));	2✔
131	}
132	return memoryBlock;	2✔
133	}
134	return null;	×
135	}
136
137	public synchronized void freeMemory(String taskId, MemoryBlock memoryBlock) {
138	if (taskIdToMemoryBlockMap.containsKey(taskId)) {	2✔
139	taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);	2✔
140	}
141	if (!memoryBlock.isFreedStatus()) {	2✔
142	getMemoryAllocator(memoryBlock.getMemoryType()).free(memoryBlock);	2✔
143	memoryUsed -= memoryBlock.size();	2✔
144	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	2✔
145	if (LOGGER.isDebugEnabled()) {	2✔
146	LOGGER.debug(String.format(	2✔
147	"Freeing working memory block (%s) with size: %d, current available memory is: %d",
148	memoryBlock.toString(), memoryBlock.size(), totalMemory - memoryUsed));	2✔
149	}
150	}
151	}	2✔
152
153	public synchronized void freeMemoryAll(String taskId) {
154	Set<MemoryBlock> memoryBlockSet = null;	1✔
155	memoryBlockSet = taskIdToMemoryBlockMap.remove(taskId);	1✔
156	long occuppiedMemory = 0;	1✔
157	if (null != memoryBlockSet) {	1✔
158	Iterator<MemoryBlock> iterator = memoryBlockSet.iterator();	1✔
159	MemoryBlock memoryBlock = null;	1✔
160	while (iterator.hasNext()) {	1✔
161	memoryBlock = iterator.next();	1✔
162	if (!memoryBlock.isFreedStatus()) {	1✔
163	occuppiedMemory += memoryBlock.size();	1✔
164	getMemoryAllocator(memoryBlock.getMemoryType()).free(memoryBlock);	1✔
165	}
166	}
167	}
168	memoryUsed -= occuppiedMemory;	1✔
169	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	1✔
170	if (LOGGER.isDebugEnabled()) {	1✔
171	LOGGER.debug(String.format(	1✔
172	"Freeing working memory of size %d. Current available memory is %d",
173	occuppiedMemory, totalMemory - memoryUsed));	1✔
174	}
175	LOGGER.info(String.format(	1✔
176	"Total working memory used after task %s is %d. Current running tasks are %s",
177	taskId, memoryUsed, StringUtils.join(taskIdToMemoryBlockMap.keySet(), ", ")));	1✔
178	}	1✔
179
180	public synchronized boolean isMemoryAvailable() {
181	return memoryUsed > totalMemory;	×
182	}
183
184	public long getUsableMemory() {
185	return totalMemory;	1✔
186	}
187
188	/**
189	* It tries to allocate memory of `size` bytes, keep retry until it allocates successfully.
190	*/
191	public static MemoryBlock allocateMemoryWithRetry(String taskId, long size)
192	throws MemoryException {
193	return allocateMemoryWithRetry(INSTANCE.memoryType, taskId, size);	2✔
194	}
195
196	public static MemoryBlock allocateMemoryWithRetry(MemoryType memoryType, String taskId,
197	long size) throws MemoryException {
198	MemoryBlock baseBlock = null;	2✔
199	int tries = 0;	2✔
200	while (tries < 300) {	2✔
201	baseBlock = INSTANCE.allocateMemory(memoryType, taskId, size);	2✔
202	if (baseBlock == null) {	2✔
203	try {
204	LOGGER.info("Working memory is not available, retry after 500 ms");	×
205	Thread.sleep(500);	×
206	} catch (InterruptedException e) {	×
207	throw new MemoryException(e);	×
208	}	×
209	} else {
210	break;
211	}
212	tries++;	×
213	}
214	if (baseBlock == null) {	2✔
215	INSTANCE.printCurrentMemoryUsage();	×
216	throw new MemoryException("Not enough working memory, please increase "	×
217	+ CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB);
218	}
219	return baseBlock;	2✔
220	}
221
222	private MemoryAllocator getMemoryAllocator(MemoryType memoryType) {
223	switch (memoryType) {	2✔
224	case ONHEAP:
225	return MemoryAllocator.HEAP;	1✔
226	default:
227	return MemoryAllocator.UNSAFE;	2✔
228	}
229	}
230
231	public static boolean isOffHeap() {
232	return offHeap;	1✔
233	}
234
235	private synchronized void printCurrentMemoryUsage() {
236	LOGGER.info(String.format("Working memory used %d, running tasks are %s",	×
237	memoryUsed, StringUtils.join(taskIdToMemoryBlockMap.keySet(), ", ")));	×
238	}	×
239	}

apache / carbondata / 1232

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