milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enable jemalloc (#18349) 

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
pull/18355/head
bigsheeper 2022-07-20 22:22:31 +08:00 committed by GitHub
parent 5b3de709f7
commit cef8b1e7cc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 7 additions and 315 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
configs/milvus.yaml
View File

@ -322,5 +322,3 @@ common:
# tls mode values [0, 1, 2]
# 0 is close, 1 is one-way authentication, 2 is two-way authentication.
tlsMode: 0
mem_purge_ratio: 0.2 # in Linux os, if memory-fragmentation-size >= used-memory * ${mem_purge_ratio}, then do `malloc_trim`

1
internal/core/src/common/CMakeLists.txt
View File

@ -15,7 +15,6 @@ set(COMMON_SRC
Schema.cpp
SystemProperty.cpp
vector_index_c.cpp
memory_c.cpp
)
add_library(milvus_common SHARED ${COMMON_SRC})

104
internal/core/src/common/memory_c.cpp
View File

@ -1,104 +0,0 @@
// Licensed to the LF AI & Data foundation 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.
#ifdef __linux__
#include <malloc.h>
#include <rapidxml/rapidxml.hpp>
#endif
#include "common/CGoHelper.h"
#include "common/memory_c.h"
#include "exceptions/EasyAssert.h"
#include "log/Log.h"
int
DoMallocTrim() {
#ifdef __linux__
/*
* The malloc_trim() function returns 1 if memory was actually released back to the system,
* or 0 if it was not possible to release any memory.
*/
return malloc_trim(0);
#endif
return 0;
}
uint64_t
ParseMallocInfo() {
#ifdef __linux__
char* mem_buffer;
size_t buffer_size;
FILE* stream;
stream = open_memstream(&mem_buffer, &buffer_size);
AssertInfo(stream, "null stream file when open_memstream");
try {
/*
* The malloc_info() function exports an XML string that describes
* the current state of the memory-allocation implementation in the caller.
* The exported XML string includes information about `fast` and `rest`.
* According to the implementation of glibc, `fast` calculates ths size of all the
* fastbins, and `rest` calculates the size of all the bins except fastbins.
* ref: <https://man7.org/linux/man-pages/man3/malloc_info.3.html>
* <https://sourceware.org/glibc/wiki/MallocInternals>
* <https://code.woboq.org/userspace/glibc/malloc/malloc.c.html#5378>
*/
auto ret = malloc_info(0, stream);
AssertInfo(ret == 0, "malloc_info failed");
fflush(stream);
rapidxml::xml_document<> doc; // character type defaults to char
doc.parse<0>(mem_buffer); // 0 means default parse flags
rapidxml::xml_node<>* malloc_root_node = doc.first_node();
AssertInfo(malloc_root_node, "null malloc_root_node detected when ParseMallocInfo");
auto total_fast_node = malloc_root_node->first_node()->next_sibling("total");
AssertInfo(total_fast_node, "null total_fast_node detected when ParseMallocInfo");
auto total_fast_size = std::stoul(total_fast_node->first_attribute("size")->value());
auto total_rest_node = total_fast_node->next_sibling("total");
AssertInfo(total_fast_node, "null total_rest_node detected when ParseMallocInfo");
auto total_rest_size = std::stoul(total_rest_node->first_attribute("size")->value());
fclose(stream);
free(mem_buffer);
return total_fast_size + total_rest_size;
} catch (std::exception& e) {
fclose(stream);
free(mem_buffer);
throw e;
}
#else
return 0; // malloc_trim is unnecessary
#endif
}
CStatus
PurgeMemory(uint64_t max_bins_size, int32_t* res) {
try {
auto fast_and_rest_total = ParseMallocInfo();
if (fast_and_rest_total >= max_bins_size) {
LOG_SEGCORE_DEBUG_ << "Purge memory fragmentation, max_bins_size(bytes) = " << max_bins_size
<< ", fast_and_rest_total(bytes) = " << fast_and_rest_total;
*res = DoMallocTrim();
} else {
*res = 0;
}
return milvus::SuccessCStatus();
} catch (std::exception& e) {
*res = 0;
return milvus::FailureCStatus(UnexpectedError, e.what());
}
}

36
internal/core/src/common/memory_c.h
View File

@ -1,36 +0,0 @@
// Licensed to the LF AI & Data foundation 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.
#pragma once
#ifdef __cplusplus
extern "C" {
#endif
#include "common/type_c.h"
/*
* In glibc, free chunks are stored in various lists based on size and history,
* so that the library can quickly find suitable chunks to satisfy allocation requests.
* The lists, called "bins".
* ref: <https://sourceware.org/glibc/wiki/MallocInternals>
*/
CStatus
PurgeMemory(uint64_t max_bins_size, int32_t* res);
#ifdef __cplusplus
}
#endif

3
internal/core/src/indexbuilder/index_c.cpp
View File

@ -50,9 +50,6 @@ DeleteIndex(CIndex index) {
AssertInfo(index, "failed to delete index, passed index was null");
auto cIndex = reinterpret_cast<milvus::indexbuilder::IndexCreatorBase*>(index);
delete cIndex;
#ifdef __linux__
malloc_trim(0);
#endif
status.error_code = Success;
status.error_msg = "";
} catch (std::exception& e) {

3
internal/core/src/segcore/collection_c.cpp
View File

@ -28,9 +28,6 @@ void
DeleteCollection(CCollection collection) {
auto col = (milvus::segcore::Collection*)collection;
delete col;
#ifdef __linux__
malloc_trim(0);
#endif
}
const char*

12
internal/core/thirdparty/CMakeLists.txt vendored
View File

@ -71,11 +71,11 @@ if ( LINUX OR APPLE)
endif()
# ******************************* Thridparty jemalloc ********************************
#if ( LINUX )
# add_subdirectory( jemalloc )
#endif()
if ( LINUX )
add_subdirectory( jemalloc )
endif()
# ******************************* Thridparty rapidxml ********************************
if ( LINUX )
add_subdirectory( rapidxml )
endif()
#if ( LINUX )
# add_subdirectory( rapidxml )
#endif()

8
internal/core/thirdparty/jemalloc/CMakeLists.txt vendored
View File

@ -46,13 +46,7 @@ endif ()
list(APPEND
JEMALLOC_CONFIGURE_COMMAND
"--prefix=${JEMALLOC_PREFIX}"
"--libdir=${JEMALLOC_LIB_DIR}"
"--with-jemalloc-prefix=je_milvus_"
"--with-private-namespace=je_milvus_private_"
"--without-export"
# See https://github.com/jemalloc/jemalloc/issues/1237
"--disable-initial-exec-tls"
${EP_LOG_OPTIONS})
"--libdir=${JEMALLOC_LIB_DIR}")
if (CMAKE_BUILD_TYPE EQUAL "DEBUG")
# Enable jemalloc debug checks when Milvus itself has debugging enabled
list(APPEND JEMALLOC_CONFIGURE_COMMAND "--enable-debug")

32
internal/core/unittest/test_c_api.cpp
View File

@ -19,7 +19,6 @@
#include <unordered_set>
#include "common/LoadInfo.h"
#include "common/memory_c.h"
#include "knowhere/index/VecIndexFactory.h"
#include "knowhere/index/vector_index/IndexIVFPQ.h"
#include "knowhere/index/vector_index/helpers/IndexParameter.h"
@ -259,37 +258,6 @@ serialize(const Message* msg) {
return ret;
}
#ifdef __linux__
// TEST(Common, Memory_benchmark) {
// auto run_times = 1000000;
// auto start = std::chrono::high_resolution_clock::now();
//
// int32_t res;
// for (int i = 0; i < run_times; i++) {
// PurgeMemory(UINT64_MAX /*never malloc_trim*/, &res);
// assert(res == 0);
// }
//
// auto stop = std::chrono::high_resolution_clock::now();
// auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stop - start);
//
// std::cout << run_times << " times taken by PurgeMemory: " << duration.count() << " milliseconds" << std::endl;
// // 1000000 times taken by PurgeMemory: 8307 milliseconds
//}
TEST(Common, Memory) {
int32_t res;
auto status = PurgeMemory(UINT64_MAX /*never malloc_trim*/, &res);
assert(res == 0);
assert(status.error_code == Success);
status = PurgeMemory(0 /*do malloc_trim*/, &res);
assert(res == 1);
assert(status.error_code == Success);
}
#endif
TEST(CApiTest, InsertTest) {
auto c_collection = NewCollection(get_default_schema_config());
auto segment = NewSegment(c_collection, Growing, -1);

25
internal/querynode/reduce.go
View File

@ -25,11 +25,6 @@ package querynode
import "C"
import (
"fmt"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/log"
memutil "github.com/milvus-io/milvus/internal/util/memutil"
metricsinfo "github.com/milvus-io/milvus/internal/util/metricsinfo"
)
type sliceInfo struct {
@ -139,26 +134,6 @@ func deleteSearchResultDataBlobs(cSearchResultDataBlobs searchResultDataBlobs) {
C.DeleteSearchResultDataBlobs(cSearchResultDataBlobs)
}
func purgeMemoryAfterReduce() {
// try to do a purgeMemory operation after reduce
usedMem := metricsinfo.GetUsedMemoryCount()
if usedMem == 0 {
log.Error("Get 0 usedMemory when deleteSearchResultDataBlobs, which is unexpected")
return
}
maxBinsSize := uint64(float64(usedMem) * Params.CommonCfg.MemPurgeRatio)
released, err := memutil.PurgeMemory(maxBinsSize)
if err != nil {
log.Error(err.Error())
}
if released {
log.Info("purge memory done, memory was released back to the system successfully",
zap.Uint64("usedMem", usedMem),
zap.Float64("memPurgeRatio", Params.CommonCfg.MemPurgeRatio),
zap.Uint64("maxBinsSize", maxBinsSize))
}
}
func deleteSearchResults(results []*SearchResult) {
if len(results) == 0 {
return

9
internal/querynode/reduce_test.go
View File

@ -104,15 +104,6 @@ func TestReduce_AllFunc(t *testing.T) {
deleteCollection(collection)
}
func TestReduce_deleteSearchResultDataBlobs(t *testing.T) {
t.Run("test purge memory", func(t *testing.T) {
bak := Params.CommonCfg.MemPurgeRatio
Params.CommonCfg.MemPurgeRatio = 0
deleteSearchResultDataBlobs(nil)
Params.CommonCfg.MemPurgeRatio = bak
})
}
func TestReduce_Invalid(t *testing.T) {
t.Run("nil plan", func(t *testing.T) {
plan := &SearchPlan{}

2
internal/querynode/task_search.go
View File

@ -120,7 +120,6 @@ func (s *searchTask) searchOnStreaming() error {
zap.Int64("collectionID", s.CollectionID), zap.Error(sErr))
return sErr
}
defer purgeMemoryAfterReduce()
defer deleteSearchResults(partResults)
return s.reduceResults(searchReq, partResults)
}
@ -156,7 +155,6 @@ func (s *searchTask) searchOnHistorical() error {
if err != nil {
return err
}
defer purgeMemoryAfterReduce()
defer deleteSearchResults(partResults)
return s.reduceResults(searchReq, partResults)
}

40
internal/util/memutil/purge_memory.go
View File

@ -1,40 +0,0 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// 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 metricsinfo
/*
#cgo pkg-config: milvus_common milvus_segcore
#include <stdlib.h>
#include "common/vector_index_c.h"
#include "common/memory_c.h"
*/
import "C"
import (
"fmt"
"unsafe"
)
func PurgeMemory(maxBinsSize uint64) (bool, error) {
var res C.int32_t
cMaxBinsSize := C.uint64_t(maxBinsSize)
status := C.PurgeMemory(cMaxBinsSize, &res)
if status.error_code == 0 {
return int32(res) > 0, nil
}
defer C.free(unsafe.Pointer(status.error_msg))
errorMsg := string(C.GoString(status.error_msg))
errorCode := int32(status.error_code)
return false, fmt.Errorf("PurgeMemory failed, errorCode = %d, errorMsg = %s", errorCode, errorMsg)
}

39
internal/util/memutil/purge_memory_test.go
View File

@ -1,39 +0,0 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// 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 metricsinfo
import (
"math"
"testing"
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
)
func TestPurgeMemory(t *testing.T) {
usedMem := metricsinfo.GetUsedMemoryCount()
assert.True(t, usedMem > 0)
maxBinsSize := uint64(float64(usedMem) * 0.2)
_, err := PurgeMemory(maxBinsSize)
assert.NoError(t, err)
// do not malloc_trim
res, err := PurgeMemory(math.MaxUint64)
assert.NoError(t, err)
assert.False(t, res)
// do malloc_trim
res, err = PurgeMemory(0)
assert.NoError(t, err)
assert.True(t, res)
}

6
internal/util/paramtable/component_param.go
View File

@ -131,7 +131,6 @@ type commonConfig struct {
SimdType string
AuthorizationEnabled bool
MemPurgeRatio float64
}
func (p *commonConfig) init(base *BaseTable) {
@ -169,7 +168,6 @@ func (p *commonConfig) init(base *BaseTable) {
p.initStorageType()
p.initEnableAuthorization()
p.initMemoryPurgeRatio()
}
func (p *commonConfig) initClusterPrefix() {
@ -373,10 +371,6 @@ func (p *commonConfig) initEnableAuthorization() {
p.AuthorizationEnabled = p.Base.ParseBool("common.security.authorizationEnabled", false)
}
func (p *commonConfig) initMemoryPurgeRatio() {
p.MemPurgeRatio = p.Base.ParseFloatWithDefault("common.mem_purge_ratio", 0.2)
}
///////////////////////////////////////////////////////////////////////////////
// --- rootcoord ---
type rootCoordConfig struct {