admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-08 01:58:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'branch-0.5.0' into caiyd_refactor_config_1

Browse Source 
Former-commit-id: 28ba312753b541b8fde198345943f974cfb6a452
This commit is contained in:
yudong.cai 2019-09-25 20:09:40 +08:00
commit 44b539728c
119 changed files with 7987 additions and 852 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ci/jenkinsfile/cluster_deploy2dev.groovy
View File

@ -5,7 +5,7 @@ try {
dir ("milvus-helm") {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
dir ("milvus/milvus-cluster") {
sh "helm install --wait --timeout 300 --set roServers.image.tag=${DOCKER_VERSION} --set woServers.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP -f ci/values.yaml --name ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster --namespace milvus-cluster --version 0.4.0 . "
sh "helm install --wait --timeout 300 --set roServers.image.tag=${DOCKER_VERSION} --set woServers.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP -f ci/values.yaml --name ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster --namespace milvus-cluster --version 0.5.0 . "
}
}
/*

2
ci/jenkinsfile/deploy2dev.groovy
View File

@ -5,7 +5,7 @@ try {
dir ("milvus-helm") {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
dir ("milvus/milvus-gpu") {
sh "helm install --wait --timeout 300 --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/values.yaml --namespace milvus-1 --version 0.4.0 ."
sh "helm install --wait --timeout 300 --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/values.yaml --namespace milvus-1 --version 0.5.0 ."
}
}
} catch (exc) {

2
ci/jenkinsfile/deploy2staging.groovy
View File

@ -5,7 +5,7 @@ try {
dir ("milvus-helm") {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
dir ("milvus/milvus-gpu") {
sh "helm install --wait --timeout 300 --set engine.image.repository=\"zilliz.azurecr.cn/milvus/engine\" --set engine.image.tag=${DOCKER_VERSION} --set expose.type=loadBalancer --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/values.yaml --namespace milvus-1 --version 0.4.0 ."
sh "helm install --wait --timeout 300 --set engine.image.repository=\"zilliz.azurecr.cn/milvus/engine\" --set engine.image.tag=${DOCKER_VERSION} --set expose.type=loadBalancer --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/values.yaml --namespace milvus-1 --version 0.5.0 ."
}
}
} catch (exc) {

2
ci/jenkinsfile/dev_test.groovy
View File

@ -16,7 +16,7 @@ timeout(time: 30, unit: 'MINUTES') {
}
dir ("milvus-helm") {
dir ("milvus/milvus-gpu") {
sh "helm install --wait --timeout 300 --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/db_backend/mysql_values.yaml --namespace milvus-2 --version 0.4.0 ."
sh "helm install --wait --timeout 300 --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/db_backend/mysql_values.yaml --namespace milvus-2 --version 0.5.0 ."
}
}
dir ("${PROJECT_NAME}_test") {

11
ci/jenkinsfile/milvus_build.groovy
View File

@ -5,17 +5,12 @@ container('milvus-build-env') {
try {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'SubmoduleOption',disableSubmodules: false,parentCredentials: true,recursiveSubmodules: true,reference: '',trackingSubmodules: false]], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
/*
dir ("cpp/thirdparty/knowhere") {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'SubmoduleOption',disableSubmodules: false,parentCredentials: true,recursiveSubmodules: true,reference: '',trackingSubmodules: false]], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/knowhere.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
sh "./build.sh -t ${params.BUILD_TYPE} -p ${knowhere_build_dir} -j"
}
*/
dir ("cpp") {
sh "git config --global user.email \"test@zilliz.com\""
sh "git config --global user.name \"test\""
sh "./build.sh -t ${params.BUILD_TYPE} -j -u -c"
withCredentials([usernamePassword(credentialsId: "${params.JFROG_USER}", usernameVariable: 'USERNAME', passwordVariable: 'PASSWORD')]) {
sh "export JFROG_ARTFACTORY_URL='${params.JFROG_ARTFACTORY_URL}' && export JFROG_USER_NAME='${USERNAME}' && export JFROG_PASSWORD='${PASSWORD}' && ./build.sh -t ${params.BUILD_TYPE} -j -u -c"
}
}
} catch (exc) {
updateGitlabCommitStatus name: 'Build Engine', state: 'failed'

11
ci/jenkinsfile/milvus_build_no_ut.groovy
View File

@ -5,17 +5,12 @@ container('milvus-build-env') {
try {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'SubmoduleOption',disableSubmodules: false,parentCredentials: true,recursiveSubmodules: true,reference: '',trackingSubmodules: false]], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
/*
dir ("cpp/thirdparty/knowhere") {
checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'SubmoduleOption',disableSubmodules: false,parentCredentials: true,recursiveSubmodules: true,reference: '',trackingSubmodules: false]], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/knowhere.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
sh "./build.sh -t ${params.BUILD_TYPE} -p ${knowhere_build_dir} -j"
}
*/
dir ("cpp") {
sh "git config --global user.email \"test@zilliz.com\""
sh "git config --global user.name \"test\""
sh "./build.sh -t ${params.BUILD_TYPE} -j"
withCredentials([usernamePassword(credentialsId: "${params.JFROG_USER}", usernameVariable: 'USERNAME', passwordVariable: 'PASSWORD')]) {
sh "export JFROG_ARTFACTORY_URL='${params.JFROG_ARTFACTORY_URL}' && export JFROG_USER_NAME='${USERNAME}' && export JFROG_PASSWORD='${PASSWORD}' && ./build.sh -t ${params.BUILD_TYPE} -j"
}
}
} catch (exc) {
updateGitlabCommitStatus name: 'Build Engine', state: 'failed'

2
ci/jenkinsfile/staging_test.groovy
View File

@ -17,7 +17,7 @@ timeout(time: 40, unit: 'MINUTES') {
}
dir ("milvus-helm") {
dir ("milvus/milvus-gpu") {
sh "helm install --wait --timeout 300 --set engine.image.repository=\"zilliz.azurecr.cn/milvus/engine\" --set engine.image.tag=${DOCKER_VERSION} --set expose.type=loadBalancer --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/db_backend/mysql_values.yaml --namespace milvus-2 --version 0.4.0 ."
sh "helm install --wait --timeout 300 --set engine.image.repository=\"zilliz.azurecr.cn/milvus/engine\" --set engine.image.tag=${DOCKER_VERSION} --set expose.type=loadBalancer --name ${env.JOB_NAME}-${env.BUILD_NUMBER} -f ci/db_backend/mysql_values.yaml --namespace milvus-2 --version 0.5.0 ."
}
}
dir ("${PROJECT_NAME}_test") {

5
cpp/CHANGELOG.md
View File

@ -20,6 +20,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-555 - Remove old scheduler
- MS-574 - Milvus configuration refactor
- MS-578 - Make sure milvus5.0 don't crack 0.3.1 data
- MS-585 - Update namespace in scheduler
## New Feature
@ -29,6 +30,8 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-567 - Add NOTICE.md
- MS-569 - Complete the NOTICE.md
- MS-575 - Add Clang-format & Clang-tidy & Cpplint
- MS-586 - Remove BUILD_FAISS_WITH_MKL option
- MS-590 - Refine cmake code to support cpplint
# Milvus 0.4.0 (2019-09-12)
@ -63,6 +66,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-510 - unittest out of memory and crashed
- MS-507 - Dataset 10m-512, index type sq8，performance in-normal when set CPU_CACHE to 16 or 64
- MS-543 - SearchTask fail without exception
- MS-582 - grafana displays changes frequently
## Improvement
- MS-327 - Clean code for milvus
@ -148,6 +152,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-539 - Remove old task code
- MS-546 - Add simple mode resource_config
- MS-570 - Add prometheus docker-compose file
- MS-576 - Scheduler refactor
## New Feature
- MS-343 - Implement ResourceMgr

61
cpp/CMakeLists.txt
View File

@ -43,8 +43,8 @@ endif()
set(MILVUS_VERSION "${GIT_BRANCH_NAME}")
string(REGEX MATCH "[0-9]+\\.[0-9]+\\.[0-9]" MILVUS_VERSION "${MILVUS_VERSION}")
set(CLANG_FORMAT_VERSION "6.0")
find_package(ClangTools)
set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
set(BUILD_SUPPORT_DIR "${CMAKE_SOURCE_DIR}/build-support")
if(CMAKE_BUILD_TYPE STREQUAL "Release")
@ -65,7 +65,7 @@ if(MILVUS_VERSION_MAJOR STREQUAL ""
OR MILVUS_VERSION_MINOR STREQUAL ""
OR MILVUS_VERSION_PATCH STREQUAL "")
message(WARNING "Failed to determine Milvus version from git branch name")
set(MILVUS_VERSION "0.4.0")
set(MILVUS_VERSION "0.5.0")
endif()
message(STATUS "Build version = ${MILVUS_VERSION}")
@ -173,12 +173,29 @@ if(NOT LINT_EXCLUSIONS_FILE)
set(LINT_EXCLUSIONS_FILE ${BUILD_SUPPORT_DIR}/lint_exclusions.txt)
endif()
find_program(CPPLINT_BIN NAMES cpplint cpplint.py HINTS ${BUILD_SUPPORT_DIR})
message(STATUS "Found cpplint executable at ${CPPLINT_BIN}")
#
# "make format" and "make check-format" targets
# "make lint" targets
#
add_custom_target(lint
${PYTHON_EXECUTABLE}
${BUILD_SUPPORT_DIR}/run_cpplint.py
--cpplint_binary
${CPPLINT_BIN}
--exclude_globs
${LINT_EXCLUSIONS_FILE}
--source_dir
${CMAKE_CURRENT_SOURCE_DIR}/src
${MILVUS_LINT_QUIET})
#
# "make clang-format" and "make check-clang-format" targets
#
if(${CLANG_FORMAT_FOUND})
# runs clang format and updates files in place.
add_custom_target(format
add_custom_target(clang-format
${PYTHON_EXECUTABLE}
${BUILD_SUPPORT_DIR}/run_clang_format.py
--clang_format_binary
@ -191,7 +208,7 @@ if(${CLANG_FORMAT_FOUND})
${MILVUS_LINT_QUIET})
# runs clang format and exits with a non-zero exit code if any files need to be reformatted
add_custom_target(check-format
add_custom_target(check-clang-format
${PYTHON_EXECUTABLE}
${BUILD_SUPPORT_DIR}/run_clang_format.py
--clang_format_binary
@ -203,3 +220,37 @@ if(${CLANG_FORMAT_FOUND})
${MILVUS_LINT_QUIET})
endif()
#
# "make clang-tidy" and "make check-clang-tidy" targets
#
if(${CLANG_TIDY_FOUND})
# runs clang-tidy and attempts to fix any warning automatically
add_custom_target(clang-tidy
${PYTHON_EXECUTABLE}
${BUILD_SUPPORT_DIR}/run_clang_tidy.py
--clang_tidy_binary
${CLANG_TIDY_BIN}
--exclude_globs
${LINT_EXCLUSIONS_FILE}
--compile_commands
${CMAKE_BINARY_DIR}/compile_commands.json
--source_dir
${CMAKE_CURRENT_SOURCE_DIR}/src
--fix
${MILVUS_LINT_QUIET})
# runs clang-tidy and exits with a non-zero exit code if any errors are found.
add_custom_target(check-clang-tidy
${PYTHON_EXECUTABLE}
${BUILD_SUPPORT_DIR}/run_clang_tidy.py
--clang_tidy_binary
${CLANG_TIDY_BIN}
--exclude_globs
${LINT_EXCLUSIONS_FILE}
--compile_commands
${CMAKE_BINARY_DIR}/compile_commands.json
--source_dir
${CMAKE_CURRENT_SOURCE_DIR}/src
${MILVUS_LINT_QUIET})
endif()

7
cpp/CONTRIBUTING.md
View File

@ -20,9 +20,9 @@ Before you make any contributions, make sure you follow this list.
Contributions to Milvus fall into the following categories.
1. To report a bug or a problem with documentation, please file an [issue](https://github.com/milvus-io/milvus/issues/new) providing the details of the problem. If you believe the issue needs priority attention, please comment on the issue to notify the team.
2. To propose a new feature, please file a new feature request [issue](https://github.com/milvus-io/milvus/issues/new). Describe the intended feature and discuss the design and implementation with the team and community. Once the team agrees that the plan looks good, go ahead and implement it, following the [Contributing code].
3. To implement a feature or bug-fix for an existing outstanding issue, follow the [Contributing code]. If you need more context on a particular issue, comment on the issue to let people know.
1. To report a bug or a problem with documentation, please file an [issue](https://github.com/milvus-io/milvus/issues/new/choose) providing the details of the problem. If you believe the issue needs priority attention, please comment on the issue to notify the team.
2. To propose a new feature, please file a new feature request [issue](https://github.com/milvus-io/milvus/issues/new/choose). Describe the intended feature and discuss the design and implementation with the team and community. Once the team agrees that the plan looks good, go ahead and implement it, following the [Contributing code](CONTRIBUTING.md#contributing-code).
3. To implement a feature or bug-fix for an existing outstanding issue, follow the [Contributing code](CONTRIBUTING.md#contributing-code). If you need more context on a particular issue, comment on the issue to let people know.
## How can I contribute?
@ -44,6 +44,7 @@ Before sending your pull requests for review, make sure your changes are consist
## Coding Style
## Run unit test
```shell

21
cpp/README.md
View File

@ -25,7 +25,7 @@ Keep up-to-date with newest releases and latest updates by reading [Releases](ht
The data is stored and computed on a distributed architecture. This lets you scale data sizes up and down without redesigning the system.
## Architecture
![Milvus_arch](https://www.milvus-io/docs/master/assets/milvus_arch.png)
![Milvus_arch](https://milvus.io/docs/assets/milvus_arch.png)
## Get started
@ -44,10 +44,12 @@ Use Docker to install Milvus is a breeze. See the [Milvus install guide](https:/
```shell
# Install tools
Centos7 :
$ yum install gfortran qt4 flex bison mysql-devel mysql
$ yum install gfortran qt4 flex bison
$ yum install mysql-devel mysql
Ubuntu16.04 :
$ sudo apt-get install gfortran qt4-qmake flex bison libmysqlclient-dev mysql-client
$ sudo apt-get install gfortran qt4-qmake flex bison
$ sudo apt-get install libmysqlclient-dev mysql-client
```
@ -96,12 +98,21 @@ please reinstall CMake with curl:
$ sudo make install
```
##### code format and linting
```shell
CentOS 7:
$ yum install clang
Ubuntu 16.04:
$ sudo apt-get install clang-format clang-tidy
$ ./build.sh -l
```
##### Run unit test
```shell
$ ./build.sh -u
or
$ ./build.sh --unittest
```
##### Run code coverage

6476
cpp/build-support/cpplint.py vendored Executable file
View File

File diff suppressed because it is too large Load Diff

6
cpp/build-support/lint_exclusions.txt
View File

@ -0,0 +1,6 @@
*cmake-build-debug*
*cmake-build-release*
*cmake_build*
*src/thirdparty*
*src/core/thirdparty*
*src/grpc*

126
cpp/build-support/run_clang_tidy.py Executable file
View File

@ -0,0 +1,126 @@
#!/usr/bin/env python
# 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.
from __future__ import print_function
import argparse
import multiprocessing as mp
import lintutils
from subprocess import PIPE
import sys
from functools import partial
def _get_chunk_key(filenames):
# lists are not hashable so key on the first filename in a chunk
return filenames[0]
# clang-tidy outputs complaints in '/path:line_number: complaint' format,
# so we can scan its output to get a list of files to fix
def _check_some_files(completed_processes, filenames):
result = completed_processes[_get_chunk_key(filenames)]
return lintutils.stdout_pathcolonline(result, filenames)
def _check_all(cmd, filenames):
# each clang-tidy instance will process 16 files
chunks = lintutils.chunk(filenames, 16)
cmds = [cmd + some for some in chunks]
results = lintutils.run_parallel(cmds, stderr=PIPE, stdout=PIPE)
error = False
# record completed processes (keyed by the first filename in the input
# chunk) for lookup in _check_some_files
completed_processes = {
_get_chunk_key(some): result
for some, result in zip(chunks, results)
}
checker = partial(_check_some_files, completed_processes)
pool = mp.Pool()
try:
# check output of completed clang-tidy invocations in parallel
for problem_files, stdout in pool.imap(checker, chunks):
if problem_files:
msg = "clang-tidy suggested fixes for {}"
print("\n".join(map(msg.format, problem_files)))
print(stdout)
error = True
except Exception:
error = True
raise
finally:
pool.terminate()
pool.join()
if error:
sys.exit(1)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Runs clang-tidy on all ")
parser.add_argument("--clang_tidy_binary",
required=True,
help="Path to the clang-tidy binary")
parser.add_argument("--exclude_globs",
help="Filename containing globs for files "
"that should be excluded from the checks")
parser.add_argument("--compile_commands",
required=True,
help="compile_commands.json to pass clang-tidy")
parser.add_argument("--source_dir",
required=True,
help="Root directory of the source code")
parser.add_argument("--fix", default=False,
action="store_true",
help="If specified, will attempt to fix the "
"source code instead of recommending fixes, "
"defaults to %(default)s")
parser.add_argument("--quiet", default=False,
action="store_true",
help="If specified, only print errors")
arguments = parser.parse_args()
exclude_globs = []
if arguments.exclude_globs:
for line in open(arguments.exclude_globs):
exclude_globs.append(line.strip())
linted_filenames = []
for path in lintutils.get_sources(arguments.source_dir, exclude_globs):
linted_filenames.append(path)
if not arguments.quiet:
msg = 'Tidying {}' if arguments.fix else 'Checking {}'
print("\n".join(map(msg.format, linted_filenames)))
cmd = [
arguments.clang_tidy_binary,
'-p',
arguments.compile_commands
]
if arguments.fix:
cmd.append('-fix')
results = lintutils.run_parallel(
[cmd + some for some in lintutils.chunk(linted_filenames, 16)])
for returncode, stdout, stderr in results:
if returncode != 0:
sys.exit(returncode)
else:
_check_all(cmd, linted_filenames)

132
cpp/build-support/run_cpplint.py Executable file
View File

@ -0,0 +1,132 @@
#!/usr/bin/env python
# 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.
from __future__ import print_function
import lintutils
from subprocess import PIPE, STDOUT
import argparse
import multiprocessing as mp
import sys
import platform
from functools import partial
# NOTE(wesm):
#
# * readability/casting is disabled as it aggressively warns about functions
# with names like "int32", so "int32(x)", where int32 is a function name,
# warns with
_filters = '''
-whitespace/comments
-readability/casting
-readability/todo
-readability/alt_tokens
-build/header_guard
-build/c++11
-runtime/references
-build/include_order
'''.split()
def _get_chunk_key(filenames):
# lists are not hashable so key on the first filename in a chunk
return filenames[0]
def _check_some_files(completed_processes, filenames):
# cpplint outputs complaints in '/path:line_number: complaint' format,
# so we can scan its output to get a list of files to fix
result = completed_processes[_get_chunk_key(filenames)]
return lintutils.stdout_pathcolonline(result, filenames)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Runs cpplint on all of the source files.")
parser.add_argument("--cpplint_binary",
required=True,
help="Path to the cpplint binary")
parser.add_argument("--exclude_globs",
help="Filename containing globs for files "
"that should be excluded from the checks")
parser.add_argument("--source_dir",
required=True,
help="Root directory of the source code")
parser.add_argument("--quiet", default=False,
action="store_true",
help="If specified, only print errors")
arguments = parser.parse_args()
exclude_globs = []
if arguments.exclude_globs:
for line in open(arguments.exclude_globs):
exclude_globs.append(line.strip())
linted_filenames = []
for path in lintutils.get_sources(arguments.source_dir, exclude_globs):
linted_filenames.append(str(path))
cmd = [
arguments.cpplint_binary,
'--verbose=2',
'--linelength=90',
'--filter=' + ','.join(_filters)
]
if (arguments.cpplint_binary.endswith('.py') and
platform.system() == 'Windows'):
# Windows doesn't support executable scripts; execute with
# sys.executable
cmd.insert(0, sys.executable)
if arguments.quiet:
cmd.append('--quiet')
else:
print("\n".join(map(lambda x: "Linting {}".format(x),
linted_filenames)))
# lint files in chunks: each invocation of cpplint will process 16 files
chunks = lintutils.chunk(linted_filenames, 16)
cmds = [cmd + some for some in chunks]
results = lintutils.run_parallel(cmds, stdout=PIPE, stderr=STDOUT)
error = False
# record completed processes (keyed by the first filename in the input
# chunk) for lookup in _check_some_files
completed_processes = {
_get_chunk_key(filenames): result
for filenames, result in zip(chunks, results)
}
checker = partial(_check_some_files, completed_processes)
pool = mp.Pool()
try:
# scan the outputs of various cpplint invocations in parallel to
# distill a list of problematic files
for problem_files, stdout in pool.imap(checker, chunks):
if problem_files:
if isinstance(stdout, bytes):
stdout = stdout.decode('utf8')
print(stdout, file=sys.stderr)
error = True
except Exception:
error = True
raise
finally:
pool.terminate()
pool.join()
sys.exit(1 if error else 0)

69
cpp/build.sh
View File

@ -7,12 +7,19 @@ MAKE_CLEAN="OFF"
BUILD_COVERAGE="OFF"
DB_PATH="/opt/milvus"
PROFILING="OFF"
BUILD_FAISS_WITH_MKL="OFF"
USE_JFROG_CACHE="OFF"
RUN_CPPLINT="OFF"
CUDA_COMPILER=/usr/local/cuda/bin/nvcc
while getopts "p:d:t:uhrcgmj" arg
while getopts "p:d:t:ulrcgjh" arg
do
case $arg in
p)
INSTALL_PREFIX=$OPTARG
;;
d)
DB_PATH=$OPTARG
;;
t)
BUILD_TYPE=$OPTARG # BUILD_TYPE
;;
@ -20,11 +27,8 @@ do
echo "Build and run unittest cases" ;
BUILD_UNITTEST="ON";
;;
p)
INSTALL_PREFIX=$OPTARG
;;
d)
DB_PATH=$OPTARG
l)
RUN_CPPLINT="ON"
;;
r)
if [[ -d cmake_build ]]; then
@ -38,9 +42,6 @@ do
g)
PROFILING="ON"
;;
m)
BUILD_FAISS_WITH_MKL="ON"
;;
j)
USE_JFROG_CACHE="ON"
;;
@ -48,23 +49,24 @@ do
echo "
parameter:
-t: build type(default: Debug)
-u: building unit test options(default: OFF)
-p: install prefix(default: $(pwd)/milvus)
-d: db path(default: /opt/milvus)
-t: build type(default: Debug)
-u: building unit test options(default: OFF)
-l: run cpplint, clang-format and clang-tidy(default: OFF)
-r: remove previous build directory(default: OFF)
-c: code coverage(default: OFF)
-g: profiling(default: OFF)
-m: build faiss with MKL(default: OFF)
-j: use jfrog cache build directory
-j: use jfrog cache build directory(default: OFF)
-h: help
usage:
./build.sh -t \${BUILD_TYPE} [-u] [-h] [-g] [-r] [-c] [-k] [-m] [-j]
./build.sh -p \${INSTALL_PREFIX} -t \${BUILD_TYPE} [-u] [-l] [-r] [-c] [-g] [-j] [-h]
"
exit 0
;;
?)
echo "unknown argument"
echo "ERROR! unknown argument"
exit 1
;;
esac
@ -72,40 +74,61 @@ done
if [[ ! -d cmake_build ]]; then
mkdir cmake_build
MAKE_CLEAN="ON"
fi
cd cmake_build
CUDA_COMPILER=/usr/local/cuda/bin/nvcc
if [[ ${MAKE_CLEAN} == "ON" ]]; then
CMAKE_CMD="cmake -DBUILD_UNIT_TEST=${BUILD_UNITTEST} \
CMAKE_CMD="cmake \
-DBUILD_UNIT_TEST=${BUILD_UNITTEST} \
-DCMAKE_INSTALL_PREFIX=${INSTALL_PREFIX}
-DCMAKE_BUILD_TYPE=${BUILD_TYPE} \
-DCMAKE_CUDA_COMPILER=${CUDA_COMPILER} \
-DBUILD_COVERAGE=${BUILD_COVERAGE} \
-DMILVUS_DB_PATH=${DB_PATH} \
-DMILVUS_ENABLE_PROFILING=${PROFILING} \
-DBUILD_FAISS_WITH_MKL=${BUILD_FAISS_WITH_MKL} \
-DUSE_JFROG_CACHE=${USE_JFROG_CACHE} \
../"
echo ${CMAKE_CMD}
${CMAKE_CMD}
if [[ ${MAKE_CLEAN} == "ON" ]]; then
make clean
fi
if [[ ${RUN_CPPLINT} == "ON" ]]; then
# cpplint check
make lint
if [ $? -ne 0 ]; then
echo "ERROR! cpplint check not pass"
exit 1
fi
# clang-format check
make check-clang-format
if [ $? -ne 0 ]; then
echo "ERROR! clang-format check failed"
exit 1
fi
# clang-tidy check
make check-clang-tidy
if [ $? -ne 0 ]; then
echo "ERROR! clang-tidy check failed"
exit 1
fi
else
# compile and build
make -j 4 || exit 1
# strip binary symbol
if [[ ${BUILD_TYPE} != "Debug" ]]; then
strip src/milvus_server
fi
make install || exit 1
# evaluate code coverage
if [[ ${BUILD_COVERAGE} == "ON" ]]; then
cd -
bash `pwd`/coverage.sh
cd -
fi
fi

1
cpp/cmake/DefineOptions.cmake
View File

@ -110,6 +110,7 @@ endif()
#----------------------------------------------------------------------
set_option_category("Test and benchmark")
unset(MILVUS_BUILD_TESTS CACHE)
if (BUILD_UNIT_TEST)
define_option(MILVUS_BUILD_TESTS "Build the MILVUS googletest unit tests" ON)
else()

31
cpp/cmake/ThirdPartyPackages.cmake
View File

@ -137,9 +137,26 @@ if(NOT DEFINED USE_JFROG_CACHE)
set(USE_JFROG_CACHE "OFF")
endif()
if(USE_JFROG_CACHE STREQUAL "ON")
set(JFROG_ARTFACTORY_CACHE_URL "http://192.168.1.201:80/artifactory/generic-local/milvus/thirdparty/cache/${CMAKE_OS_NAME}/${MILVUS_BUILD_ARCH}/${BUILD_TYPE}")
set(JFROG_USER_NAME "test")
set(JFROG_PASSWORD "Fantast1c")
if(DEFINED ENV{JFROG_ARTFACTORY_URL})
set(JFROG_ARTFACTORY_URL "$ENV{JFROG_ARTFACTORY_URL}")
endif()
if(NOT DEFINED JFROG_ARTFACTORY_URL)
message(FATAL_ERROR "JFROG_ARTFACTORY_URL is not set")
endif()
set(JFROG_ARTFACTORY_CACHE_URL "${JFROG_ARTFACTORY_URL}/generic-local/milvus/thirdparty/cache/${CMAKE_OS_NAME}/${MILVUS_BUILD_ARCH}/${BUILD_TYPE}")
if(DEFINED ENV{JFROG_USER_NAME})
set(JFROG_USER_NAME "$ENV{JFROG_USER_NAME}")
endif()
if(NOT DEFINED JFROG_USER_NAME)
message(FATAL_ERROR "JFROG_USER_NAME is not set")
endif()
if(DEFINED ENV{JFROG_PASSWORD})
set(JFROG_PASSWORD "$ENV{JFROG_PASSWORD}")
endif()
if(NOT DEFINED JFROG_PASSWORD)
message(FATAL_ERROR "JFROG_PASSWORD is not set")
endif()
set(THIRDPARTY_PACKAGE_CACHE "${THIRDPARTY_DIR}/cache")
endif()
@ -242,14 +259,6 @@ foreach(_VERSION_ENTRY ${TOOLCHAIN_VERSIONS_TXT})
set(${_LIB_NAME} "${_LIB_VERSION}")
endforeach()
if(DEFINED ENV{MILVUS_ARROW_URL})
set(ARROW_SOURCE_URL "$ENV{MILVUS_ARROW_URL}")
else()
set(ARROW_SOURCE_URL
"https://github.com/youny626/arrow.git"
)
endif()
if(DEFINED ENV{MILVUS_BOOST_URL})
set(BOOST_SOURCE_URL "$ENV{MILVUS_BOOST_URL}")
else()

62
cpp/src/cache/LRU.h vendored
View File

@ -34,78 +34,78 @@ public:
typedef typename std::list<key_value_pair_t>::iterator list_iterator_t;
typedef typename std::list<key_value_pair_t>::reverse_iterator reverse_list_iterator_t;
LRU(size_t max_size) : _max_size(max_size) {}
LRU(size_t max_size) : max_size_(max_size) {}
void put(const key_t& key, const value_t& value) {
auto it = _cache_items_map.find(key);
_cache_items_list.push_front(key_value_pair_t(key, value));
if (it != _cache_items_map.end()) {
_cache_items_list.erase(it->second);
_cache_items_map.erase(it);
auto it = cache_items_map_.find(key);
cache_items_list_.push_front(key_value_pair_t(key, value));
if (it != cache_items_map_.end()) {
cache_items_list_.erase(it->second);
cache_items_map_.erase(it);
}
_cache_items_map[key] = _cache_items_list.begin();
cache_items_map_[key] = cache_items_list_.begin();
if (_cache_items_map.size() > _max_size) {
auto last = _cache_items_list.end();
if (cache_items_map_.size() > max_size_) {
auto last = cache_items_list_.end();
last--;
_cache_items_map.erase(last->first);
_cache_items_list.pop_back();
cache_items_map_.erase(last->first);
cache_items_list_.pop_back();
}
}
const value_t& get(const key_t& key) {
auto it = _cache_items_map.find(key);
if (it == _cache_items_map.end()) {
auto it = cache_items_map_.find(key);
if (it == cache_items_map_.end()) {
throw std::range_error("There is no such key in cache");
} else {
_cache_items_list.splice(_cache_items_list.begin(), _cache_items_list, it->second);
cache_items_list_.splice(cache_items_list_.begin(), cache_items_list_, it->second);
return it->second->second;
}
}
void erase(const key_t& key) {
auto it = _cache_items_map.find(key);
if (it != _cache_items_map.end()) {
_cache_items_list.erase(it->second);
_cache_items_map.erase(it);
auto it = cache_items_map_.find(key);
if (it != cache_items_map_.end()) {
cache_items_list_.erase(it->second);
cache_items_map_.erase(it);
}
}
bool exists(const key_t& key) const {
return _cache_items_map.find(key) != _cache_items_map.end();
return cache_items_map_.find(key) != cache_items_map_.end();
}
size_t size() const {
return _cache_items_map.size();
return cache_items_map_.size();
}
list_iterator_t begin() {
_iter = _cache_items_list.begin();
return _iter;
iter_ = cache_items_list_.begin();
return iter_;
}
list_iterator_t end() {
return _cache_items_list.end();
return cache_items_list_.end();
}
reverse_list_iterator_t rbegin() {
return _cache_items_list.rbegin();
return cache_items_list_.rbegin();
}
reverse_list_iterator_t rend() {
return _cache_items_list.rend();
return cache_items_list_.rend();
}
void clear() {
_cache_items_list.clear();
_cache_items_map.clear();
cache_items_list_.clear();
cache_items_map_.clear();
}
private:
std::list<key_value_pair_t> _cache_items_list;
std::unordered_map<key_t, list_iterator_t> _cache_items_map;
size_t _max_size;
list_iterator_t _iter;
std::list<key_value_pair_t> cache_items_list_;
std::unordered_map<key_t, list_iterator_t> cache_items_map_;
size_t max_size_;
list_iterator_t iter_;
};
} // cache

4
cpp/src/core/cmake/ThirdPartyPackagesCore.cmake
View File

@ -123,9 +123,7 @@ if(NOT DEFINED USE_JFROG_CACHE)
set(USE_JFROG_CACHE "OFF")
endif()
if(USE_JFROG_CACHE STREQUAL "ON")
set(JFROG_ARTFACTORY_CACHE_URL "http://192.168.1.201:80/artifactory/generic-local/milvus/thirdparty/cache/${CMAKE_OS_NAME}/${KNOWHERE_BUILD_ARCH}/${BUILD_TYPE}")
set(JFROG_USER_NAME "test")
set(JFROG_PASSWORD "Fantast1c")
set(JFROG_ARTFACTORY_CACHE_URL "${JFROG_ARTFACTORY_URL}/generic-local/milvus/thirdparty/cache/${CMAKE_OS_NAME}/${KNOWHERE_BUILD_ARCH}/${BUILD_TYPE}")
set(THIRDPARTY_PACKAGE_CACHE "${THIRDPARTY_DIR}/cache")
endif()

2
cpp/src/db/DBFactory.cpp
View File

@ -36,7 +36,7 @@ namespace engine {
DBOptions DBFactory::BuildOption() {
auto meta = MetaFactory::BuildOption();
DBOptions options;
options.meta = meta;
options.meta_ = meta;
return options;
}

14
cpp/src/db/DBImpl.cpp
View File

@ -56,7 +56,7 @@ DBImpl::DBImpl(const DBOptions& options)
shutting_down_(true),
compact_thread_pool_(1, 1),
index_thread_pool_(1, 1) {
meta_ptr_ = MetaFactory::Build(options.meta, options.mode);
meta_ptr_ = MetaFactory::Build(options.meta_, options.mode_);
mem_mgr_ = MemManagerFactory::Build(meta_ptr_, options_);
Start();
}
@ -77,7 +77,7 @@ Status DBImpl::Start() {
shutting_down_.store(false, std::memory_order_release);
//for distribute version, some nodes are read only
if (options_.mode != DBOptions::MODE::READ_ONLY) {
if (options_.mode_ != DBOptions::MODE::READ_ONLY) {
ENGINE_LOG_TRACE << "StartTimerTasks";
bg_timer_thread_ = std::thread(&DBImpl::BackgroundTimerTask, this);
}
@ -98,7 +98,7 @@ Status DBImpl::Stop() {
//wait compaction/buildindex finish
bg_timer_thread_.join();
if (options_.mode != DBOptions::MODE::READ_ONLY) {
if (options_.mode_ != DBOptions::MODE::READ_ONLY) {
meta_ptr_->CleanUp();
}
@ -133,9 +133,9 @@ Status DBImpl::DeleteTable(const std::string& table_id, const meta::DatesT& date
meta_ptr_->DeleteTable(table_id); //soft delete table
//scheduler will determine when to delete table files
auto nres = ResMgrInst::GetInstance()->GetNumOfComputeResource();
auto nres = scheduler::ResMgrInst::GetInstance()->GetNumOfComputeResource();
scheduler::DeleteJobPtr job = std::make_shared<scheduler::DeleteJob>(0, table_id, meta_ptr_, nres);
JobMgrInst::GetInstance()->Put(job);
scheduler::JobMgrInst::GetInstance()->Put(job);
job->WaitAndDelete();
} else {
meta_ptr_->DropPartitionsByDates(table_id, dates);
@ -649,7 +649,7 @@ Status DBImpl::BackgroundMergeFiles(const std::string& table_id) {
bool has_merge = false;
for (auto& kv : raw_files) {
auto files = kv.second;
if (files.size() < options_.merge_trigger_number) {
if (files.size() < options_.merge_trigger_number_) {
ENGINE_LOG_DEBUG << "Files number not greater equal than merge trigger number, skip merge action";
continue;
}
@ -684,7 +684,7 @@ void DBImpl::BackgroundCompaction(std::set<std::string> table_ids) {
meta_ptr_->Archive();
int ttl = 5*meta::M_SEC;//default: file will be deleted after 5 minutes
if (options_.mode == DBOptions::MODE::CLUSTER) {
if (options_.mode_ == DBOptions::MODE::CLUSTER) {
ttl = meta::D_SEC;
}
meta_ptr_->CleanUpFilesWithTTL(ttl);

16
cpp/src/db/Options.h
View File

@ -52,10 +52,10 @@ private:
};
struct DBMetaOptions {
std::string path;
std::vector<std::string> slave_paths;
std::string backend_uri;
ArchiveConf archive_conf = ArchiveConf("delete");
std::string path_;
std::vector<std::string> slave_paths_;
std::string backend_uri_;
ArchiveConf archive_conf_ = ArchiveConf("delete");
}; // DBMetaOptions
struct DBOptions {
@ -65,11 +65,11 @@ struct DBOptions {
READ_ONLY
} MODE;
uint16_t merge_trigger_number = 2;
DBMetaOptions meta;
int mode = MODE::SINGLE;
uint16_t merge_trigger_number_ = 2;
DBMetaOptions meta_;
int mode_ = MODE::SINGLE;
size_t insert_buffer_size = 4 * ONE_GB;
size_t insert_buffer_size_ = 4 * ONE_GB;
bool insert_cache_immediately_ = false;
}; // Options

56
cpp/src/db/Utils.cpp
View File

@ -21,6 +21,7 @@
#include <mutex>
#include <chrono>
#include <regex>
#include <boost/filesystem.hpp>
namespace zilliz {
@ -42,8 +43,8 @@ std::string ConstructParentFolder(const std::string& db_path, const meta::TableF
}
std::string GetTableFileParentFolder(const DBMetaOptions& options, const meta::TableFileSchema& table_file) {
uint64_t path_count = options.slave_paths.size() + 1;
std::string target_path = options.path;
uint64_t path_count = options.slave_paths_.size() + 1;
std::string target_path = options.path_;
uint64_t index = 0;
if(meta::TableFileSchema::NEW_INDEX == table_file.file_type_) {
@ -60,7 +61,7 @@ std::string GetTableFileParentFolder(const DBMetaOptions& options, const meta::T
}
if (index > 0) {
target_path = options.slave_paths[index - 1];
target_path = options.slave_paths_[index - 1];
}
return ConstructParentFolder(target_path, table_file);
@ -77,7 +78,7 @@ long GetMicroSecTimeStamp() {
}
Status CreateTablePath(const DBMetaOptions& options, const std::string& table_id) {
std::string db_path = options.path;
std::string db_path = options.path_;
std::string table_path = db_path + TABLES_FOLDER + table_id;
auto status = server::CommonUtil::CreateDirectory(table_path);
if (!status.ok()) {
@ -85,7 +86,7 @@ Status CreateTablePath(const DBMetaOptions& options, const std::string& table_id
return status;
}
for(auto& path : options.slave_paths) {
for(auto& path : options.slave_paths_) {
table_path = path + TABLES_FOLDER + table_id;
status = server::CommonUtil::CreateDirectory(table_path);
if (!status.ok()) {
@ -98,8 +99,8 @@ Status CreateTablePath(const DBMetaOptions& options, const std::string& table_id
}
Status DeleteTablePath(const DBMetaOptions& options, const std::string& table_id, bool force) {
std::vector<std::string> paths = options.slave_paths;
paths.push_back(options.path);
std::vector<std::string> paths = options.slave_paths_;
paths.push_back(options.path_);
for(auto& path : paths) {
std::string table_path = path + TABLES_FOLDER + table_id;
@ -131,13 +132,13 @@ Status CreateTableFilePath(const DBMetaOptions& options, meta::TableFileSchema&
}
Status GetTableFilePath(const DBMetaOptions& options, meta::TableFileSchema& table_file) {
std::string parent_path = ConstructParentFolder(options.path, table_file);
std::string parent_path = ConstructParentFolder(options.path_, table_file);
std::string file_path = parent_path + "/" + table_file.file_id_;
if(boost::filesystem::exists(file_path)) {
table_file.location_ = file_path;
return Status::OK();
} else {
for(auto& path : options.slave_paths) {
for(auto& path : options.slave_paths_) {
parent_path = ConstructParentFolder(path, table_file);
file_path = parent_path + "/" + table_file.file_id_;
if(boost::filesystem::exists(file_path)) {
@ -148,7 +149,7 @@ Status GetTableFilePath(const DBMetaOptions& options, meta::TableFileSchema& tab
}
std::string msg = "Table file doesn't exist: " + file_path;
ENGINE_LOG_ERROR << msg << " in path: " << options.path
ENGINE_LOG_ERROR << msg << " in path: " << options.path_
<< " for table: " << table_file.table_id_;
return Status(DB_ERROR, msg);
@ -195,6 +196,41 @@ meta::DateT GetDate() {
return GetDate(std::time(nullptr), 0);
}
// URI format: dialect://username:password@host:port/database
Status ParseMetaUri(const std::string& uri, MetaUriInfo& info) {
std::string dialect_regex = "(.*)";
std::string username_tegex = "(.*)";
std::string password_regex = "(.*)";
std::string host_regex = "(.*)";
std::string port_regex = "(.*)";
std::string db_name_regex = "(.*)";
std::string uri_regex_str =
dialect_regex + "\\:\\/\\/" +
username_tegex + "\\:" +
password_regex + "\\@" +
host_regex + "\\:" +
port_regex + "\\/" +
db_name_regex;
std::regex uri_regex(uri_regex_str);
std::smatch pieces_match;
if (std::regex_match(uri, pieces_match, uri_regex)) {
info.dialect_ = pieces_match[1].str();
info.username_ = pieces_match[2].str();
info.password_ = pieces_match[3].str();
info.host_ = pieces_match[4].str();
info.port_ = pieces_match[5].str();
info.db_name_ = pieces_match[6].str();
//TODO: verify host, port...
} else {
return Status(DB_INVALID_META_URI, "Invalid meta uri: " + uri);
}
return Status::OK();
}
} // namespace utils
} // namespace engine
} // namespace milvus

11
cpp/src/db/Utils.h
View File

@ -44,6 +44,17 @@ meta::DateT GetDate(const std::time_t &t, int day_delta = 0);
meta::DateT GetDate();
meta::DateT GetDateWithDelta(int day_delta);
struct MetaUriInfo {
std::string dialect_;
std::string username_;
std::string password_;
std::string host_;
std::string port_;
std::string db_name_;
};
Status ParseMetaUri(const std::string& uri, MetaUriInfo& info);
} // namespace utils
} // namespace engine
} // namespace milvus

32
cpp/src/db/engine/ExecutionEngineImpl.cpp
View File

@ -100,11 +100,8 @@ VecIndexPtr ExecutionEngineImpl::CreatetVecIndex(EngineType type) {
}
Status ExecutionEngineImpl::AddWithIds(long n, const float *xdata, const long *xids) {
auto ec = index_->Add(n, xdata, xids);
if (ec != KNOWHERE_SUCCESS) {
return Status(DB_ERROR, "Add error");
}
return Status::OK();
auto status = index_->Add(n, xdata, xids);
return status;
}
size_t ExecutionEngineImpl::Count() const {
@ -132,11 +129,8 @@ size_t ExecutionEngineImpl::PhysicalSize() const {
}
Status ExecutionEngineImpl::Serialize() {
auto ec = write_index(index_, location_);
if (ec != KNOWHERE_SUCCESS) {
return Status(DB_ERROR, "Serialize: write to disk error");
}
return Status::OK();
auto status = write_index(index_, location_);
return status;
}
Status ExecutionEngineImpl::Load(bool to_cache) {
@ -255,12 +249,11 @@ Status ExecutionEngineImpl::Merge(const std::string &location) {
}
if (auto file_index = std::dynamic_pointer_cast<BFIndex>(to_merge)) {
auto ec = index_->Add(file_index->Count(), file_index->GetRawVectors(), file_index->GetRawIds());
if (ec != KNOWHERE_SUCCESS) {
auto status = index_->Add(file_index->Count(), file_index->GetRawVectors(), file_index->GetRawIds());
if (!status.ok()) {
ENGINE_LOG_ERROR << "Merge: Add Error";
return Status(DB_ERROR, "Merge: Add Error");
}
return Status::OK();
return status;
} else {
return Status(DB_ERROR, "file index type is not idmap");
}
@ -288,11 +281,11 @@ ExecutionEngineImpl::BuildIndex(const std::string &location, EngineType engine_t
build_cfg["nlist"] = nlist_;
AutoGenParams(to_index->GetType(), Count(), build_cfg);
auto ec = to_index->BuildAll(Count(),
auto status = to_index->BuildAll(Count(),
from_index->GetRawVectors(),
from_index->GetRawIds(),
build_cfg);
if (ec != KNOWHERE_SUCCESS) { throw Exception(DB_ERROR, "Build index error"); }
if (!status.ok()) { throw Exception(DB_ERROR, status.message()); }
return std::make_shared<ExecutionEngineImpl>(to_index, location, engine_type, metric_type_, nlist_);
}
@ -310,12 +303,11 @@ Status ExecutionEngineImpl::Search(long n,
ENGINE_LOG_DEBUG << "Search Params: [k] " << k << " [nprobe] " << nprobe;
auto cfg = Config::object{{"k", k}, {"nprobe", nprobe}};
auto ec = index_->Search(n, data, distances, labels, cfg);
if (ec != KNOWHERE_SUCCESS) {
auto status = index_->Search(n, data, distances, labels, cfg);
if (!status.ok()) {
ENGINE_LOG_ERROR << "Search error";
return Status(DB_ERROR, "Search: Search Error");
}
return Status::OK();
return status;
}
Status ExecutionEngineImpl::Cache() {

2
cpp/src/db/insert/MemManagerImpl.cpp
View File

@ -43,7 +43,7 @@ Status MemManagerImpl::InsertVectors(const std::string &table_id_,
const float *vectors_,
IDNumbers &vector_ids_) {
while (GetCurrentMem() > options_.insert_buffer_size) {
while (GetCurrentMem() > options_.insert_buffer_size_) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}

40
cpp/src/db/meta/MetaFactory.cpp
View File

@ -20,13 +20,14 @@
#include "MySQLMetaImpl.h"
#include "utils/Log.h"
#include "utils/Exception.h"
#include "db/Utils.h"
#include <stdlib.h>
#include <time.h>
#include <sstream>
#include <cstdlib>
#include <string>
#include <regex>
#include <string.h>
namespace zilliz {
namespace milvus {
@ -42,45 +43,30 @@ namespace engine {
}
DBMetaOptions meta;
meta.path = p;
meta.path_ = p;
return meta;
}
meta::MetaPtr MetaFactory::Build(const DBMetaOptions &metaOptions, const int &mode) {
std::string uri = metaOptions.backend_uri;
std::string uri = metaOptions.backend_uri_;
std::string dialectRegex = "(.*)";
std::string usernameRegex = "(.*)";
std::string passwordRegex = "(.*)";
std::string hostRegex = "(.*)";
std::string portRegex = "(.*)";
std::string dbNameRegex = "(.*)";
std::string uriRegexStr = dialectRegex + "\\:\\/\\/" +
usernameRegex + "\\:" +
passwordRegex + "\\@" +
hostRegex + "\\:" +
portRegex + "\\/" +
dbNameRegex;
std::regex uriRegex(uriRegexStr);
std::smatch pieces_match;
utils::MetaUriInfo uri_info;
auto status = utils::ParseMetaUri(uri, uri_info);
if(!status.ok()) {
ENGINE_LOG_ERROR << "Wrong URI format: URI = " << uri;
throw InvalidArgumentException("Wrong URI format ");
}
if (std::regex_match(uri, pieces_match, uriRegex)) {
std::string dialect = pieces_match[1].str();
std::transform(dialect.begin(), dialect.end(), dialect.begin(), ::tolower);
if (dialect.find("mysql") != std::string::npos) {
if (strcasecmp(uri_info.dialect_.c_str(), "mysql") == 0) {
ENGINE_LOG_INFO << "Using MySQL";
return std::make_shared<meta::MySQLMetaImpl>(metaOptions, mode);
} else if (dialect.find("sqlite") != std::string::npos) {
} else if (strcasecmp(uri_info.dialect_.c_str(), "sqlite") == 0) {
ENGINE_LOG_INFO << "Using SQLite";
return std::make_shared<meta::SqliteMetaImpl>(metaOptions);
} else {
ENGINE_LOG_ERROR << "Invalid dialect in URI: dialect = " << dialect;
ENGINE_LOG_ERROR << "Invalid dialect in URI: dialect = " << uri_info.dialect_;
throw InvalidArgumentException("URI dialect is not mysql / sqlite");
}
} else {
ENGINE_LOG_ERROR << "Wrong URI format: URI = " << uri;
throw InvalidArgumentException("Wrong URI format ");
}
}
} // namespace engine

317
cpp/src/db/meta/MySQLMetaImpl.cpp
View File

@ -19,21 +19,22 @@
#include "db/IDGenerator.h"
#include "db/Utils.h"
#include "utils/Log.h"
#include "utils/Exception.h"
#include "MetaConsts.h"
#include "metrics/Metrics.h"
#include <unistd.h>
#include <sstream>
#include <iostream>
#include <boost/filesystem.hpp>
#include <chrono>
#include <fstream>
#include <regex>
#include <string>
#include <mutex>
#include <thread>
#include "mysql++/mysql++.h"
#include <string.h>
#include <boost/filesystem.hpp>
#include <mysql++/mysql++.h>
namespace zilliz {
@ -56,8 +57,112 @@ Status HandleException(const std::string &desc, const char* what = nullptr) {
}
}
class MetaField {
public:
MetaField(const std::string& name, const std::string& type, const std::string& setting)
: name_(name),
type_(type),
setting_(setting) {
}
std::string name() const {
return name_;
}
std::string ToString() const {
return name_ + " " + type_ + " " + setting_;
}
// mysql field type has additional information. for instance, a filed type is defined as 'BIGINT'
// we get the type from sql is 'bigint(20)', so we need to ignore the '(20)'
bool IsEqual(const MetaField& field) const {
size_t name_len_min = field.name_.length() > name_.length() ? name_.length() : field.name_.length();
size_t type_len_min = field.type_.length() > type_.length() ? type_.length() : field.type_.length();
return strncasecmp(field.name_.c_str(), name_.c_str(), name_len_min) == 0 &&
strncasecmp(field.type_.c_str(), type_.c_str(), type_len_min) == 0;
}
private:
std::string name_;
std::string type_;
std::string setting_;
};
using MetaFields = std::vector<MetaField>;
class MetaSchema {
public:
MetaSchema(const std::string& name, const MetaFields& fields)
: name_(name),
fields_(fields) {
}
std::string name() const {
return name_;
}
std::string ToString() const {
std::string result;
for(auto& field : fields_) {
if(!result.empty()) {
result += ",";
}
result += field.ToString();
}
return result;
}
//if the outer fields contains all this MetaSchema fields, return true
//otherwise return false
bool IsEqual(const MetaFields& fields) const {
std::vector<std::string> found_field;
for(const auto& this_field : fields_) {
for(const auto& outer_field : fields) {
if(this_field.IsEqual(outer_field)) {
found_field.push_back(this_field.name());
break;
}
}
}
return found_field.size() == fields_.size();
}
private:
std::string name_;
MetaFields fields_;
};
//Tables schema
static const MetaSchema TABLES_SCHEMA(META_TABLES, {
MetaField("id", "BIGINT", "PRIMARY KEY AUTO_INCREMENT"),
MetaField("table_id", "VARCHAR(255)", "UNIQUE NOT NULL"),
MetaField("state", "INT", "NOT NULL"),
MetaField("dimension", "SMALLINT", "NOT NULL"),
MetaField("created_on", "BIGINT", "NOT NULL"),
MetaField("flag", "BIGINT", "DEFAULT 0 NOT NULL"),
MetaField("index_file_size", "BIGINT", "DEFAULT 1024 NOT NULL"),
MetaField("engine_type", "INT", "DEFAULT 1 NOT NULL"),
MetaField("nlist", "INT", "DEFAULT 16384 NOT NULL"),
MetaField("metric_type", "INT", "DEFAULT 1 NOT NULL"),
});
//TableFiles schema
static const MetaSchema TABLEFILES_SCHEMA(META_TABLEFILES, {
MetaField("id", "BIGINT", "PRIMARY KEY AUTO_INCREMENT"),
MetaField("table_id", "VARCHAR(255)", "NOT NULL"),
MetaField("engine_type", "INT", "DEFAULT 1 NOT NULL"),
MetaField("file_id", "VARCHAR(255)", "NOT NULL"),
MetaField("file_type", "INT", "DEFAULT 0 NOT NULL"),
MetaField("file_size", "BIGINT", "DEFAULT 0 NOT NULL"),
MetaField("row_count", "BIGINT", "DEFAULT 0 NOT NULL"),
MetaField("updated_time", "BIGINT", "NOT NULL"),
MetaField("created_on", "BIGINT", "NOT NULL"),
MetaField("date", "INT", "DEFAULT -1 NOT NULL"),
});
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
MySQLMetaImpl::MySQLMetaImpl(const DBMetaOptions &options_, const int &mode)
: options_(options_),
mode_(mode) {
@ -84,55 +189,99 @@ Status MySQLMetaImpl::NextFileId(std::string &file_id) {
return Status::OK();
}
void MySQLMetaImpl::ValidateMetaSchema() {
if(nullptr == mysql_connection_pool_) {
return;
}
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return;
}
auto validate_func = [&](const MetaSchema& schema) {
Query query_statement = connectionPtr->query();
query_statement << "DESC " << schema.name() << ";";
MetaFields exist_fields;
try {
StoreQueryResult res = query_statement.store();
for (size_t i = 0; i < res.num_rows(); i++) {
const Row &row = res[i];
std::string name, type;
row["Field"].to_string(name);
row["Type"].to_string(type);
exist_fields.push_back(MetaField(name, type, ""));
}
} catch (std::exception &e) {
ENGINE_LOG_DEBUG << "Meta table '" << schema.name() << "' not exist and will be created";
}
if(exist_fields.empty()) {
return true;
}
return schema.IsEqual(exist_fields);
};
//verify Tables
if (!validate_func(TABLES_SCHEMA)) {
throw Exception(DB_INCOMPATIB_META, "Meta Tables schema is created by Milvus old version");
}
//verufy TableFiles
if (!validate_func(TABLEFILES_SCHEMA)) {
throw Exception(DB_INCOMPATIB_META, "Meta TableFiles schema is created by Milvus old version");
}
}
Status MySQLMetaImpl::Initialize() {
if (!boost::filesystem::is_directory(options_.path)) {
auto ret = boost::filesystem::create_directory(options_.path);
//step 1: create db root path
if (!boost::filesystem::is_directory(options_.path_)) {
auto ret = boost::filesystem::create_directory(options_.path_);
if (!ret) {
std::string msg = "Failed to create db directory " + options_.path;
std::string msg = "Failed to create db directory " + options_.path_;
ENGINE_LOG_ERROR << msg;
return Status(DB_META_TRANSACTION_FAILED, msg);
}
}
std::string uri = options_.backend_uri;
std::string uri = options_.backend_uri_;
std::string dialectRegex = "(.*)";
std::string usernameRegex = "(.*)";
std::string passwordRegex = "(.*)";
std::string hostRegex = "(.*)";
std::string portRegex = "(.*)";
std::string dbNameRegex = "(.*)";
std::string uriRegexStr = dialectRegex + "\\:\\/\\/" +
usernameRegex + "\\:" +
passwordRegex + "\\@" +
hostRegex + "\\:" +
portRegex + "\\/" +
dbNameRegex;
std::regex uriRegex(uriRegexStr);
std::smatch pieces_match;
if (std::regex_match(uri, pieces_match, uriRegex)) {
std::string dialect = pieces_match[1].str();
std::transform(dialect.begin(), dialect.end(), dialect.begin(), ::tolower);
if (dialect.find("mysql") == std::string::npos) {
return Status(DB_ERROR, "URI's dialect is not MySQL");
//step 2: parse and check meta uri
utils::MetaUriInfo uri_info;
auto status = utils::ParseMetaUri(uri, uri_info);
if(!status.ok()) {
std::string msg = "Wrong URI format: " + uri;
ENGINE_LOG_ERROR << msg;
throw Exception(DB_INVALID_META_URI, msg);
}
std::string username = pieces_match[2].str();
std::string password = pieces_match[3].str();
std::string serverAddress = pieces_match[4].str();
if (strcasecmp(uri_info.dialect_.c_str(), "mysql") != 0) {
std::string msg = "URI's dialect is not MySQL";
ENGINE_LOG_ERROR << msg;
throw Exception(DB_INVALID_META_URI, msg);
}
//step 3: connect mysql
int thread_hint = std::thread::hardware_concurrency();
int max_pool_size = (thread_hint == 0) ? 8 : thread_hint;
unsigned int port = 0;
if (!pieces_match[5].str().empty()) {
port = std::stoi(pieces_match[5].str());
if (!uri_info.port_.empty()) {
port = std::stoi(uri_info.port_);
}
std::string dbName = pieces_match[6].str();
int threadHint = std::thread::hardware_concurrency();
int maxPoolSize = threadHint == 0 ? 8 : threadHint;
mysql_connection_pool_ =
std::make_shared<MySQLConnectionPool>(dbName, username, password, serverAddress, port, maxPoolSize);
std::make_shared<MySQLConnectionPool>(uri_info.db_name_, uri_info.username_,
uri_info.password_, uri_info.host_, port, max_pool_size);
ENGINE_LOG_DEBUG << "MySQL connection pool: maximum pool size = " << std::to_string(max_pool_size);
ENGINE_LOG_DEBUG << "MySQL connection pool: maximum pool size = " << std::to_string(maxPoolSize);
//step 4: validate to avoid open old version schema
ValidateMetaSchema();
//step 5: create meta tables
try {
if (mode_ != DBOptions::MODE::READ_ONLY) {
@ -140,7 +289,7 @@ Status MySQLMetaImpl::Initialize() {
}
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -154,17 +303,7 @@ Status MySQLMetaImpl::Initialize() {
Query InitializeQuery = connectionPtr->query();
InitializeQuery << "CREATE TABLE IF NOT EXISTS " <<
META_TABLES << " " <<
"(id BIGINT PRIMARY KEY AUTO_INCREMENT, " <<
"table_id VARCHAR(255) UNIQUE NOT NULL, " <<
"state INT NOT NULL, " <<
"dimension SMALLINT NOT NULL, " <<
"created_on BIGINT NOT NULL, " <<
"flag BIGINT DEFAULT 0 NOT NULL, " <<
"index_file_size BIGINT DEFAULT 1024 NOT NULL, " <<
"engine_type INT DEFAULT 1 NOT NULL, " <<
"nlist INT DEFAULT 16384 NOT NULL, " <<
"metric_type INT DEFAULT 1 NOT NULL);";
TABLES_SCHEMA.name() << " (" << TABLES_SCHEMA.ToString() + ");";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::Initialize: " << InitializeQuery.str();
@ -173,17 +312,7 @@ Status MySQLMetaImpl::Initialize() {
}
InitializeQuery << "CREATE TABLE IF NOT EXISTS " <<
META_TABLEFILES << " " <<
"(id BIGINT PRIMARY KEY AUTO_INCREMENT, " <<
"table_id VARCHAR(255) NOT NULL, " <<
"engine_type INT DEFAULT 1 NOT NULL, " <<
"file_id VARCHAR(255) NOT NULL, " <<
"file_type INT DEFAULT 0 NOT NULL, " <<
"file_size BIGINT DEFAULT 0 NOT NULL, " <<
"row_count BIGINT DEFAULT 0 NOT NULL, " <<
"updated_time BIGINT NOT NULL, " <<
"created_on BIGINT NOT NULL, " <<
"date INT DEFAULT -1 NOT NULL);";
TABLEFILES_SCHEMA.name() << " (" << TABLEFILES_SCHEMA.ToString() + ");";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::Initialize: " << InitializeQuery.str();
@ -195,10 +324,6 @@ Status MySQLMetaImpl::Initialize() {
} catch (std::exception &e) {
return HandleException("GENERAL ERROR DURING INITIALIZATION", e.what());
}
} else {
ENGINE_LOG_ERROR << "Wrong URI format. URI = " << uri;
return Status(DB_ERROR, "Wrong URI format");
}
return Status::OK();
}
@ -226,7 +351,7 @@ Status MySQLMetaImpl::DropPartitionsByDates(const std::string &table_id,
dateListStr = dateListStr.substr(0, dateListStr.size() - 2); //remove the last ", "
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -258,7 +383,7 @@ Status MySQLMetaImpl::CreateTable(TableSchema &table_schema) {
try {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -337,7 +462,7 @@ Status MySQLMetaImpl::FilesByType(const std::string &table_id,
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -417,7 +542,7 @@ Status MySQLMetaImpl::UpdateTableIndex(const std::string &table_id, const TableI
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -476,7 +601,7 @@ Status MySQLMetaImpl::UpdateTableFlag(const std::string &table_id, int64_t flag)
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -508,7 +633,7 @@ Status MySQLMetaImpl::DescribeTableIndex(const std::string &table_id, TableIndex
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -548,7 +673,7 @@ Status MySQLMetaImpl::DropTableIndex(const std::string &table_id) {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -611,7 +736,7 @@ Status MySQLMetaImpl::DeleteTable(const std::string &table_id) {
try {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -648,7 +773,7 @@ Status MySQLMetaImpl::DeleteTableFiles(const std::string &table_id) {
try {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -682,7 +807,7 @@ Status MySQLMetaImpl::DescribeTable(TableSchema &table_schema) {
server::MetricCollector metric;
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -735,7 +860,7 @@ Status MySQLMetaImpl::HasTable(const std::string &table_id, bool &has_or_not) {
server::MetricCollector metric;
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -770,7 +895,7 @@ Status MySQLMetaImpl::AllTables(std::vector<TableSchema> &table_schema_array) {
server::MetricCollector metric;
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -851,7 +976,7 @@ Status MySQLMetaImpl::CreateTableFile(TableFileSchema &file_schema) {
std::string date = std::to_string(file_schema.date_);
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -890,7 +1015,7 @@ Status MySQLMetaImpl::FilesToIndex(TableFilesSchema &files) {
server::MetricCollector metric;
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -974,7 +1099,7 @@ Status MySQLMetaImpl::FilesToSearch(const std::string &table_id,
server::MetricCollector metric;
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1093,7 +1218,7 @@ Status MySQLMetaImpl::FilesToMerge(const std::string &table_id,
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1184,7 +1309,7 @@ Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
try {
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1253,7 +1378,7 @@ Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
// PXU TODO: Support Swap
Status MySQLMetaImpl::Archive() {
auto &criterias = options_.archive_conf.GetCriterias();
auto &criterias = options_.archive_conf_.GetCriterias();
if (criterias.empty()) {
return Status::OK();
}
@ -1266,7 +1391,7 @@ Status MySQLMetaImpl::Archive() {
long now = utils::GetMicroSecTimeStamp();
try {
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1307,7 +1432,7 @@ Status MySQLMetaImpl::Size(uint64_t &result) {
try {
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1347,7 +1472,7 @@ Status MySQLMetaImpl::DiscardFiles(long long to_discard_size) {
server::MetricCollector metric;
bool status;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1412,7 +1537,7 @@ Status MySQLMetaImpl::UpdateTableFile(TableFileSchema &file_schema) {
try {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1480,7 +1605,7 @@ Status MySQLMetaImpl::UpdateTableFile(TableFileSchema &file_schema) {
Status MySQLMetaImpl::UpdateTableFilesToIndex(const std::string &table_id) {
try {
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1511,7 +1636,7 @@ Status MySQLMetaImpl::UpdateTableFiles(TableFilesSchema &files) {
try {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1596,7 +1721,7 @@ Status MySQLMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1668,7 +1793,7 @@ Status MySQLMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1719,7 +1844,7 @@ Status MySQLMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
server::MetricCollector metric;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1749,7 +1874,7 @@ Status MySQLMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
Status MySQLMetaImpl::CleanUp() {
try {
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1800,7 +1925,7 @@ Status MySQLMetaImpl::Count(const std::string &table_id, uint64_t &result) {
StoreQueryResult res;
{
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
@ -1836,14 +1961,14 @@ Status MySQLMetaImpl::Count(const std::string &table_id, uint64_t &result) {
Status MySQLMetaImpl::DropAll() {
try {
ENGINE_LOG_DEBUG << "Drop all mysql meta";
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab);
ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to database server");
}
Query dropTableQuery = connectionPtr->query();
dropTableQuery << "DROP TABLE IF EXISTS " << META_TABLES << ", " << META_TABLEFILES << ";";
dropTableQuery << "DROP TABLE IF EXISTS " << TABLES_SCHEMA.name() << ", " << TABLEFILES_SCHEMA.name() << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropAll: " << dropTableQuery.str();

5
cpp/src/db/meta/MySQLMetaImpl.h
View File

@ -103,13 +103,16 @@ class MySQLMetaImpl : public Meta {
Status NextFileId(std::string &file_id);
Status NextTableId(std::string &table_id);
Status DiscardFiles(long long to_discard_size);
void ValidateMetaSchema();
Status Initialize();
private:
const DBMetaOptions options_;
const int mode_;
std::shared_ptr<MySQLConnectionPool> mysql_connection_pool_;
bool safe_grab = false;
bool safe_grab_ = false;
// std::mutex connectionMutex_;
}; // DBMetaImpl

35
cpp/src/db/meta/SqliteMetaImpl.cpp
View File

@ -84,7 +84,6 @@ inline auto StoragePrototype(const std::string &path) {
using ConnectorT = decltype(StoragePrototype(""));
static std::unique_ptr<ConnectorT> ConnectorPtr;
using ConditionT = decltype(c(&TableFileSchema::id_) == 1UL);
SqliteMetaImpl::SqliteMetaImpl(const DBMetaOptions &options_)
: options_(options_) {
@ -111,28 +110,36 @@ Status SqliteMetaImpl::NextFileId(std::string &file_id) {
return Status::OK();
}
Status SqliteMetaImpl::Initialize() {
if (!boost::filesystem::is_directory(options_.path)) {
auto ret = boost::filesystem::create_directory(options_.path);
if (!ret) {
std::string msg = "Failed to create db directory " + options_.path;
ENGINE_LOG_ERROR << msg;
return Status(DB_INVALID_PATH, msg);
void SqliteMetaImpl::ValidateMetaSchema() {
if(ConnectorPtr == nullptr) {
return;
}
}
ConnectorPtr = std::make_unique<ConnectorT>(StoragePrototype(options_.path + "/meta.sqlite"));
//old meta could be recreated since schema changed, throw exception if meta schema is not compatible
auto ret = ConnectorPtr->sync_schema_simulate();
if(ret.find(META_TABLES) != ret.end()
&& sqlite_orm::sync_schema_result::dropped_and_recreated == ret[META_TABLES]) {
throw Exception(DB_INCOMPATIB_META, "Meta schema is created by Milvus old version");
throw Exception(DB_INCOMPATIB_META, "Meta Tables schema is created by Milvus old version");
}
if(ret.find(META_TABLEFILES) != ret.end()
&& sqlite_orm::sync_schema_result::dropped_and_recreated == ret[META_TABLEFILES]) {
throw Exception(DB_INCOMPATIB_META, "Meta schema is created by Milvus old version");
throw Exception(DB_INCOMPATIB_META, "Meta TableFiles schema is created by Milvus old version");
}
}
Status SqliteMetaImpl::Initialize() {
if (!boost::filesystem::is_directory(options_.path_)) {
auto ret = boost::filesystem::create_directory(options_.path_);
if (!ret) {
std::string msg = "Failed to create db directory " + options_.path_;
ENGINE_LOG_ERROR << msg;
return Status(DB_INVALID_PATH, msg);
}
}
ConnectorPtr = std::make_unique<ConnectorT>(StoragePrototype(options_.path_ + "/meta.sqlite"));
ValidateMetaSchema();
ConnectorPtr->sync_schema();
ConnectorPtr->open_forever(); // thread safe option
@ -879,7 +886,7 @@ Status SqliteMetaImpl::GetTableFiles(const std::string& table_id,
// PXU TODO: Support Swap
Status SqliteMetaImpl::Archive() {
auto &criterias = options_.archive_conf.GetCriterias();
auto &criterias = options_.archive_conf_.GetCriterias();
if (criterias.size() == 0) {
return Status::OK();
}

4
cpp/src/db/meta/SqliteMetaImpl.h
View File

@ -97,10 +97,12 @@ class SqliteMetaImpl : public Meta {
Status NextFileId(std::string &file_id);
Status NextTableId(std::string &table_id);
Status DiscardFiles(long to_discard_size);
void ValidateMetaSchema();
Status Initialize();
private:
const DBMetaOptions options_;
std::mutex meta_mutex_;
}; // DBMetaImpl

2
cpp/src/scheduler/Algorithm.cpp
View File

@ -20,7 +20,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
constexpr uint64_t MAXINT = std::numeric_limits<uint32_t >::max();

2
cpp/src/scheduler/Algorithm.h
View File

@ -24,7 +24,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
uint64_t
ShortestPath(const ResourcePtr &src,

48
cpp/src/scheduler/Definition.h Normal file
View File

@ -0,0 +1,48 @@
// 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.
#include <string>
#include <vector>
#include <list>
#include <queue>
#include <deque>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include "db/meta/MetaTypes.h"
#include "db/engine/EngineFactory.h"
#include "db/engine/ExecutionEngine.h"
namespace zilliz {
namespace milvus {
namespace scheduler {
using TableFileSchemaPtr = engine::meta::TableFileSchemaPtr;
using TableFileSchema = engine::meta::TableFileSchema;
using ExecutionEnginePtr = engine::ExecutionEnginePtr;
using EngineFactory = engine::EngineFactory;
using EngineType = engine::EngineType;
using MetricType = engine::MetricType;
}
}
}

3
cpp/src/scheduler/JobMgr.h
View File

@ -36,9 +36,6 @@ namespace zilliz {
namespace milvus {
namespace scheduler {
using engine::TaskPtr;
using engine::ResourceMgrPtr;
class JobMgr {
public:
explicit

2
cpp/src/scheduler/ResourceFactory.cpp
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::shared_ptr<Resource>
ResourceFactory::Create(const std::string &name,

2
cpp/src/scheduler/ResourceFactory.h
View File

@ -28,7 +28,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class ResourceFactory {
public:

2
cpp/src/scheduler/ResourceMgr.cpp
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
void

2
cpp/src/scheduler/ResourceMgr.h
View File

@ -30,7 +30,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class ResourceMgr {
public:

2
cpp/src/scheduler/SchedInst.cpp
View File

@ -25,7 +25,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
ResourceMgrPtr ResMgrInst::instance = nullptr;
std::mutex ResMgrInst::mutex_;

2
cpp/src/scheduler/SchedInst.h
View File

@ -27,7 +27,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class ResMgrInst {
public:

131
cpp/src/scheduler/Scheduler.cpp
View File

@ -16,7 +16,7 @@
// under the License.
#include <src/cache/GpuCacheMgr.h>
#include "src/cache/GpuCacheMgr.h"
#include "event/LoadCompletedEvent.h"
#include "Scheduler.h"
#include "action/Action.h"
@ -25,7 +25,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
Scheduler::Scheduler(ResourceMgrWPtr res_mgr)
: running_(false),
@ -33,6 +33,14 @@ Scheduler::Scheduler(ResourceMgrWPtr res_mgr)
if (auto mgr = res_mgr_.lock()) {
mgr->RegisterSubscriber(std::bind(&Scheduler::PostEvent, this, std::placeholders::_1));
}
event_register_.insert(std::make_pair(static_cast<uint64_t>(EventType::START_UP),
std::bind(&Scheduler::OnStartUp, this, std::placeholders::_1)));
event_register_.insert(std::make_pair(static_cast<uint64_t>(EventType::LOAD_COMPLETED),
std::bind(&Scheduler::OnLoadCompleted, this, std::placeholders::_1)));
event_register_.insert(std::make_pair(static_cast<uint64_t>(EventType::TASK_TABLE_UPDATED),
std::bind(&Scheduler::OnTaskTableUpdated, this, std::placeholders::_1)));
event_register_.insert(std::make_pair(static_cast<uint64_t>(EventType::FINISH_TASK),
std::bind(&Scheduler::OnFinishTask, this, std::placeholders::_1)));
}
@ -84,40 +92,8 @@ Scheduler::worker_function() {
void
Scheduler::Process(const EventPtr &event) {
switch (event->Type()) {
case EventType::START_UP: {
OnStartUp(event);
break;
}
case EventType::LOAD_COMPLETED: {
OnLoadCompleted(event);
break;
}
case EventType::FINISH_TASK: {
OnFinishTask(event);
break;
}
case EventType::TASK_TABLE_UPDATED: {
OnTaskTableUpdated(event);
break;
}
default: {
// TODO: logging
break;
}
}
}
void
Scheduler::OnStartUp(const EventPtr &event) {
if (auto resource = event->resource_.lock()) {
resource->WakeupLoader();
}
}
void
Scheduler::OnFinishTask(const EventPtr &event) {
auto process_event = event_register_.at(static_cast<int>(event->Type()));
process_event(event);
}
// TODO: refactor the function
@ -130,79 +106,11 @@ Scheduler::OnLoadCompleted(const EventPtr &event) {
auto task_table_type = load_completed_event->task_table_item_->task->label()->Type();
switch (task_table_type) {
case TaskLabelType::DEFAULT: {
if (not resource->HasExecutor() && load_completed_event->task_table_item_->Move()) {
auto task = load_completed_event->task_table_item_->task;
auto search_task = std::static_pointer_cast<XSearchTask>(task);
bool moved = false;
// to support test task, REFACTOR
if (auto index_engine = search_task->index_engine_) {
auto location = index_engine->GetLocation();
for (auto i = 0; i < res_mgr_.lock()->GetNumGpuResource(); ++i) {
auto index = zilliz::milvus::cache::GpuCacheMgr::GetInstance(i)->GetIndex(location);
if (index != nullptr) {
moved = true;
auto dest_resource = res_mgr_.lock()->GetResource(ResourceType::GPU, i);
Action::PushTaskToResource(load_completed_event->task_table_item_->task, dest_resource);
break;
}
}
}
if (not moved) {
Action::PushTaskToNeighbourRandomly(task, resource);
}
}
Action::DefaultLabelTaskScheduler(res_mgr_, resource, load_completed_event);
break;
}
case TaskLabelType::SPECIFIED_RESOURCE: {
// support next version
// auto self = event->resource_.lock();
// auto task = load_completed_event->task_table_item_->task;
//
// // if this resource is disk, assign it to smallest cost resource
// if (self->type() == ResourceType::DISK) {
// // step 1: calculate shortest path per resource, from disk to compute resource
// auto compute_resources = res_mgr_.lock()->GetComputeResources();
// std::vector<std::vector<std::string>> paths;
// std::vector<uint64_t> transport_costs;
// for (auto &res : compute_resources) {
// std::vector<std::string> path;
// uint64_t transport_cost = ShortestPath(self, res, res_mgr_.lock(), path);
// transport_costs.push_back(transport_cost);
// paths.emplace_back(path);
// }
//
// // step 2: select min cost, cost(resource) = avg_cost * task_to_do + transport_cost
// uint64_t min_cost = std::numeric_limits<uint64_t>::max();
// uint64_t min_cost_idx = 0;
// for (uint64_t i = 0; i < compute_resources.size(); ++i) {
// if (compute_resources[i]->TotalTasks() == 0) {
// min_cost_idx = i;
// break;
// }
// uint64_t cost = compute_resources[i]->TaskAvgCost() * compute_resources[i]->NumOfTaskToExec()
// + transport_costs[i];
// if (min_cost > cost) {
// min_cost = cost;
// min_cost_idx = i;
// }
// }
//
// // step 3: set path in task
// Path task_path(paths[min_cost_idx], paths[min_cost_idx].size() - 1);
// task->path() = task_path;
// }
//
// if (self->name() == task->path().Last()) {
// self->WakeupLoader();
// } else {
// auto next_res_name = task->path().Next();
// auto next_res = res_mgr_.lock()->GetResource(next_res_name);
// load_completed_event->task_table_item_->Move();
// next_res->task_table().Put(task);
// }
Action::SpecifiedResourceLabelTaskScheduler(res_mgr_, resource, load_completed_event);
break;
}
case TaskLabelType::BROADCAST: {
@ -216,6 +124,17 @@ Scheduler::OnLoadCompleted(const EventPtr &event) {
}
}
void
Scheduler::OnStartUp(const EventPtr &event) {
if (auto resource = event->resource_.lock()) {
resource->WakeupLoader();
}
}
void
Scheduler::OnFinishTask(const EventPtr &event) {
}
void
Scheduler::OnTaskTableUpdated(const EventPtr &event) {
if (auto resource = event->resource_.lock()) {

4
cpp/src/scheduler/Scheduler.h
View File

@ -30,7 +30,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
// TODO: refactor, not friendly to unittest, logical in framework code
@ -122,6 +122,8 @@ private:
private:
bool running_;
std::unordered_map<uint64_t, std::function<void(EventPtr)>> event_register_;
ResourceMgrWPtr res_mgr_;
std::queue<EventPtr> event_queue_;
std::thread worker_thread_;

4
cpp/src/scheduler/TaskCreator.cpp
View File

@ -45,7 +45,7 @@ TaskCreator::Create(const SearchJobPtr &job) {
std::vector<TaskPtr> tasks;
for (auto &index_file : job->index_files()) {
auto task = std::make_shared<XSearchTask>(index_file.second);
task->label() = std::make_shared<engine::DefaultLabel>();
task->label() = std::make_shared<DefaultLabel>();
task->job_ = job;
tasks.emplace_back(task);
}
@ -57,7 +57,7 @@ std::vector<TaskPtr>
TaskCreator::Create(const DeleteJobPtr &job) {
std::vector<TaskPtr> tasks;
auto task = std::make_shared<XDeleteTask>(job);
task->label() = std::make_shared<engine::BroadcastLabel>();
task->label() = std::make_shared<BroadcastLabel>();
task->job_ = job;
tasks.emplace_back(task);

4
cpp/src/scheduler/TaskCreator.h
View File

@ -39,10 +39,6 @@ namespace zilliz {
namespace milvus {
namespace scheduler {
using engine::TaskPtr;
using engine::XSearchTask;
using engine::XDeleteTask;
class TaskCreator {
public:
static std::vector<TaskPtr>

2
cpp/src/scheduler/TaskTable.cpp
View File

@ -27,7 +27,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::string
ToString(TaskTableItemState state) {

2
cpp/src/scheduler/TaskTable.h
View File

@ -27,7 +27,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
enum class TaskTableItemState {
INVALID,

2
cpp/src/scheduler/Utils.cpp
View File

@ -24,7 +24,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
uint64_t
get_current_timestamp() {

2
cpp/src/scheduler/Utils.h
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
uint64_t
get_current_timestamp();

12
cpp/src/scheduler/action/Action.h
View File

@ -18,11 +18,12 @@
#pragma once
#include "../resource/Resource.h"
#include "../ResourceMgr.h"
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class Action {
public:
@ -34,6 +35,15 @@ public:
static void
PushTaskToResource(const TaskPtr &task, const ResourcePtr &dest);
static void
DefaultLabelTaskScheduler(ResourceMgrWPtr res_mgr, ResourcePtr resource, std::shared_ptr<LoadCompletedEvent> event);
static void
SpecifiedResourceLabelTaskScheduler(ResourceMgrWPtr res_mgr,
ResourcePtr resource,
std::shared_ptr<LoadCompletedEvent> event);
};

82
cpp/src/scheduler/action/PushTaskToNeighbour.cpp
View File

@ -18,12 +18,14 @@
#include <list>
#include <random>
#include "../Algorithm.h"
#include "src/cache/GpuCacheMgr.h"
#include "Action.h"
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::vector<ResourcePtr>
get_neighbours(const ResourcePtr &self) {
@ -101,6 +103,84 @@ Action::PushTaskToResource(const TaskPtr& task, const ResourcePtr& dest) {
dest->task_table().Put(task);
}
void
Action::DefaultLabelTaskScheduler(ResourceMgrWPtr res_mgr,
ResourcePtr resource,
std::shared_ptr<LoadCompletedEvent> event) {
if (not resource->HasExecutor() && event->task_table_item_->Move()) {
auto task = event->task_table_item_->task;
auto search_task = std::static_pointer_cast<XSearchTask>(task);
bool moved = false;
//to support test task, REFACTOR
if (auto index_engine = search_task->index_engine_) {
auto location = index_engine->GetLocation();
for (auto i = 0; i < res_mgr.lock()->GetNumGpuResource(); ++i) {
auto index = zilliz::milvus::cache::GpuCacheMgr::GetInstance(i)->GetIndex(location);
if (index != nullptr) {
moved = true;
auto dest_resource = res_mgr.lock()->GetResource(ResourceType::GPU, i);
PushTaskToResource(event->task_table_item_->task, dest_resource);
break;
}
}
}
if (not moved) {
PushTaskToNeighbourRandomly(task, resource);
}
}
}
void
Action::SpecifiedResourceLabelTaskScheduler(ResourceMgrWPtr res_mgr,
ResourcePtr resource,
std::shared_ptr<LoadCompletedEvent> event) {
auto task = event->task_table_item_->task;
if (resource->type() == ResourceType::DISK) {
// step 1: calculate shortest path per resource, from disk to compute resource
auto compute_resources = res_mgr.lock()->GetComputeResources();
std::vector<std::vector<std::string>> paths;
std::vector<uint64_t> transport_costs;
for (auto &res : compute_resources) {
std::vector<std::string> path;
uint64_t transport_cost = ShortestPath(resource, res, res_mgr.lock(), path);
transport_costs.push_back(transport_cost);
paths.emplace_back(path);
}
// step 2: select min cost, cost(resource) = avg_cost * task_to_do + transport_cost
uint64_t min_cost = std::numeric_limits<uint64_t>::max();
uint64_t min_cost_idx = 0;
for (uint64_t i = 0; i < compute_resources.size(); ++i) {
if (compute_resources[i]->TotalTasks() == 0) {
min_cost_idx = i;
break;
}
uint64_t cost = compute_resources[i]->TaskAvgCost() * compute_resources[i]->NumOfTaskToExec()
+ transport_costs[i];
if (min_cost > cost) {
min_cost = cost;
min_cost_idx = i;
}
}
// step 3: set path in task
Path task_path(paths[min_cost_idx], paths[min_cost_idx].size() - 1);
task->path() = task_path;
}
if (resource->name() == task->path().Last()) {
resource->WakeupLoader();
} else {
auto next_res_name = task->path().Next();
auto next_res = res_mgr.lock()->GetResource(next_res_name);
event->task_table_item_->Move();
next_res->task_table().Put(task);
}
}
}
}
}

2
cpp/src/scheduler/event/Event.h
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
enum class EventType {
START_UP,

2
cpp/src/scheduler/event/EventDump.cpp
View File

@ -25,7 +25,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &operator<<(std::ostream &out, const Event &event) {
out << event.Dump();

2
cpp/src/scheduler/event/FinishTaskEvent.h
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class FinishTaskEvent : public Event {
public:

2
cpp/src/scheduler/event/LoadCompletedEvent.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class LoadCompletedEvent : public Event {
public:

2
cpp/src/scheduler/event/StartUpEvent.h
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class StartUpEvent : public Event {
public:

2
cpp/src/scheduler/event/TaskTableUpdatedEvent.h
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class TaskTableUpdatedEvent : public Event {
public:

2
cpp/src/scheduler/resource/Connection.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class Connection {
public:

2
cpp/src/scheduler/resource/CpuResource.cpp
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &operator<<(std::ostream &out, const CpuResource &resource) {
out << resource.Dump();

2
cpp/src/scheduler/resource/CpuResource.h
View File

@ -24,7 +24,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class CpuResource : public Resource {
public:

2
cpp/src/scheduler/resource/DiskResource.cpp
View File

@ -20,7 +20,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &operator<<(std::ostream &out, const DiskResource &resource) {
out << resource.Dump();

2
cpp/src/scheduler/resource/DiskResource.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class DiskResource : public Resource {
public:

2
cpp/src/scheduler/resource/GpuResource.cpp
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &operator<<(std::ostream &out, const GpuResource &resource) {
out << resource.Dump();

2
cpp/src/scheduler/resource/GpuResource.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class GpuResource : public Resource {
public:

2
cpp/src/scheduler/resource/Node.cpp
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
Node::Node() {
static std::atomic_uint_fast8_t counter(0);

2
cpp/src/scheduler/resource/Node.h
View File

@ -27,7 +27,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class Node;

2
cpp/src/scheduler/resource/Resource.cpp
View File

@ -22,7 +22,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &
operator<<(std::ostream &out, const Resource &resource) {

2
cpp/src/scheduler/resource/Resource.h
View File

@ -37,7 +37,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
// TODO(wxyu): Storage, Route, Executor
enum class ResourceType {

2
cpp/src/scheduler/resource/TestResource.cpp
View File

@ -20,7 +20,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
std::ostream &operator<<(std::ostream &out, const TestResource &resource) {
out << resource.Dump();

2
cpp/src/scheduler/resource/TestResource.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class TestResource : public Resource {
public:

2
cpp/src/scheduler/task/DeleteTask.cpp
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
XDeleteTask::XDeleteTask(const scheduler::DeleteJobPtr &delete_job)
: Task(TaskType::DeleteTask), delete_job_(delete_job) {}

2
cpp/src/scheduler/task/DeleteTask.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class XDeleteTask : public Task {
public:

2
cpp/src/scheduler/task/Path.h
View File

@ -23,7 +23,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class Path {
public:

8
cpp/src/scheduler/task/SearchTask.cpp
View File

@ -27,7 +27,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
static constexpr size_t PARALLEL_REDUCE_THRESHOLD = 10000;
static constexpr size_t PARALLEL_REDUCE_BATCH = 1000;
@ -79,8 +79,8 @@ std::mutex XSearchTask::merge_mutex_;
void
CollectFileMetrics(int file_type, size_t file_size) {
switch (file_type) {
case meta::TableFileSchema::RAW:
case meta::TableFileSchema::TO_INDEX: {
case TableFileSchema::RAW:
case TableFileSchema::TO_INDEX: {
server::Metrics::GetInstance().RawFileSizeHistogramObserve(file_size);
server::Metrics::GetInstance().RawFileSizeTotalIncrement(file_size);
server::Metrics::GetInstance().RawFileSizeGaugeSet(file_size);
@ -95,7 +95,7 @@ CollectFileMetrics(int file_type, size_t file_size) {
}
}
XSearchTask::XSearchTask(meta::TableFileSchemaPtr file)
XSearchTask::XSearchTask(TableFileSchemaPtr file)
: Task(TaskType::SearchTask), file_(file) {
if (file_) {
index_engine_ = EngineFactory::Build(file_->dimension_,

8
cpp/src/scheduler/task/SearchTask.h
View File

@ -18,19 +18,19 @@
#pragma once
#include "Task.h"
#include "db/meta/MetaTypes.h"
#include "scheduler/job/SearchJob.h"
#include "scheduler/Definition.h"
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
// TODO: rewrite
class XSearchTask : public Task {
public:
explicit
XSearchTask(meta::TableFileSchemaPtr file);
XSearchTask(TableFileSchemaPtr file);
void
Load(LoadType type, uint8_t device_id) override;
@ -56,7 +56,7 @@ public:
scheduler::ResultSet &result_target);
public:
meta::TableFileSchemaPtr file_;
TableFileSchemaPtr file_;
size_t index_id_ = 0;
int index_type_ = 0;

2
cpp/src/scheduler/task/Task.h
View File

@ -28,7 +28,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
enum class LoadType {
DISK2CPU,

5
cpp/src/scheduler/task/TestTask.cpp
View File

@ -22,10 +22,9 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
TestTask::TestTask(meta::TableFileSchemaPtr &file) : XSearchTask(file) {}
TestTask::TestTask(TableFileSchemaPtr &file) : XSearchTask(file) {}
void
TestTask::Load(LoadType type, uint8_t device_id) {

4
cpp/src/scheduler/task/TestTask.h
View File

@ -22,12 +22,12 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class TestTask : public XSearchTask {
public:
explicit
TestTask(meta::TableFileSchemaPtr& file);
TestTask(TableFileSchemaPtr& file);
public:
void

2
cpp/src/scheduler/tasklabel/BroadcastLabel.h
View File

@ -24,7 +24,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class BroadcastLabel : public TaskLabel {

2
cpp/src/scheduler/tasklabel/DefaultLabel.h
View File

@ -24,7 +24,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class DefaultLabel : public TaskLabel {
public:

2
cpp/src/scheduler/tasklabel/SpecResLabel.h
View File

@ -29,7 +29,7 @@ using ResourceWPtr = std::weak_ptr<Resource>;
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
class SpecResLabel : public TaskLabel {
public:

2
cpp/src/scheduler/tasklabel/TaskLabel.h
View File

@ -21,7 +21,7 @@
namespace zilliz {
namespace milvus {
namespace engine {
namespace scheduler {
enum class TaskLabelType {
DEFAULT, // means can be executed in any resource

2
cpp/src/sdk/examples/grpcsimple/src/ClientTest.cpp
View File

@ -39,7 +39,7 @@ constexpr int64_t BATCH_ROW_COUNT = 100000;
constexpr int64_t NQ = 5;
constexpr int64_t TOP_K = 10;
constexpr int64_t SEARCH_TARGET = 5000; //change this value, result is different
constexpr int64_t ADD_VECTOR_LOOP = 10;
constexpr int64_t ADD_VECTOR_LOOP = 1;
constexpr int64_t SECONDS_EACH_HOUR = 3600;
#define BLOCK_SPLITER std::cout << "===========================================" << std::endl;

20
cpp/src/server/DBWrapper.cpp
View File

@ -40,20 +40,20 @@ Status DBWrapper::StartService() {
//db config
engine::DBOptions opt;
s = config.GetDBConfigBackendUrl(opt.meta.backend_uri);
s = config.GetDBConfigBackendUrl(opt.meta_.backend_uri_);
if (!s.ok()) return s;
std::string path;
s = config.GetDBConfigPath(path);
if (!s.ok()) return s;
opt.meta.path = path + "/db";
opt.meta_.path_ = path + "/db";
std::string db_slave_path;
s = config.GetDBConfigSlavePath(db_slave_path);
if (!s.ok()) return s;
StringHelpFunctions::SplitStringByDelimeter(db_slave_path, ";", opt.meta.slave_paths);
StringHelpFunctions::SplitStringByDelimeter(db_slave_path, ";", opt.meta_.slave_paths_);
// cache config
s = config.GetCacheConfigCacheInsertData(opt.insert_cache_immediately_);
@ -64,13 +64,13 @@ Status DBWrapper::StartService() {
if (!s.ok()) return s;
if (mode == "single") {
opt.mode = engine::DBOptions::MODE::SINGLE;
opt.mode_ = engine::DBOptions::MODE::SINGLE;
}
else if (mode == "cluster") {
opt.mode = engine::DBOptions::MODE::CLUSTER;
opt.mode_ = engine::DBOptions::MODE::CLUSTER;
}
else if (mode == "read_only") {
opt.mode = engine::DBOptions::MODE::READ_ONLY;
opt.mode_ = engine::DBOptions::MODE::READ_ONLY;
}
else {
std::cerr << "ERROR: mode specified in server_config is not one of ['single', 'cluster', 'read_only']" << std::endl;
@ -112,16 +112,16 @@ Status DBWrapper::StartService() {
if (days > 0) {
criterial[engine::ARCHIVE_CONF_DAYS] = days;
}
opt.meta.archive_conf.SetCriterias(criterial);
opt.meta_.archive_conf_.SetCriterias(criterial);
//create db root folder
Status status = CommonUtil::CreateDirectory(opt.meta.path);
Status status = CommonUtil::CreateDirectory(opt.meta_.path_);
if(!status.ok()) {
std::cerr << "ERROR! Failed to create database root path: " << opt.meta.path << std::endl;
std::cerr << "ERROR! Failed to create database root path: " << opt.meta_.path_ << std::endl;
kill(0, SIGUSR1);
}
for(auto& path : opt.meta.slave_paths) {
for(auto& path : opt.meta_.slave_paths_) {
status = CommonUtil::CreateDirectory(path);
if(!status.ok()) {
std::cerr << "ERROR! Failed to create database slave path: " << path << std::endl;

18
cpp/src/server/Server.cpp
View File

@ -128,18 +128,18 @@ Server::Daemonize() {
stderr = fopen("/dev/null", "w+");
// Try to write PID of daemon to lockfile
if (!pid_filename_.empty()) {
pid_fd = open(pid_filename_.c_str(), O_RDWR | O_CREAT, 0640);
if (pid_fd < 0) {
pid_fd_ = open(pid_filename_.c_str(), O_RDWR | O_CREAT, 0640);
if (pid_fd_ < 0) {
std::cerr << "Can't open filename: " + pid_filename_ + ", Error: " + strerror(errno);
exit(EXIT_FAILURE);
}
if (lockf(pid_fd, F_TLOCK, 0) < 0) {
if (lockf(pid_fd_, F_TLOCK, 0) < 0) {
std::cerr << "Can't lock filename: " + pid_filename_ + ", Error: " + strerror(errno);
exit(EXIT_FAILURE);
}
std::string pid_file_context = std::to_string(getpid());
ssize_t res = write(pid_fd, pid_file_context.c_str(), pid_file_context.size());
ssize_t res = write(pid_fd_, pid_file_context.c_str(), pid_file_context.size());
if (res != 0) {
return;
}
@ -205,13 +205,13 @@ Server::Stop() {
std::cerr << "Milvus server is going to shutdown ..." << std::endl;
/* Unlock and close lockfile */
if (pid_fd != -1) {
int ret = lockf(pid_fd, F_ULOCK, 0);
if (pid_fd_ != -1) {
int ret = lockf(pid_fd_, F_ULOCK, 0);
if (ret != 0) {
std::cerr << "Can't lock file: " << strerror(errno) << std::endl;
exit(0);
}
ret = close(pid_fd);
ret = close(pid_fd_);
if (ret != 0) {
std::cerr << "Can't close file: " << strerror(errno) << std::endl;
exit(0);
@ -247,7 +247,7 @@ Server::LoadConfig() {
void
Server::StartService() {
engine::KnowhereResource::Initialize();
engine::StartSchedulerService();
scheduler::StartSchedulerService();
DBWrapper::GetInstance().StartService();
grpc::GrpcServer::GetInstance().Start();
}
@ -256,7 +256,7 @@ void
Server::StopService() {
grpc::GrpcServer::GetInstance().Stop();
DBWrapper::GetInstance().StopService();
engine::StopSchedulerService();
scheduler::StopSchedulerService();
engine::KnowhereResource::Finalize();
}

2
cpp/src/server/Server.h
View File

@ -52,7 +52,7 @@ class Server {
private:
int64_t daemonized_ = 0;
int pid_fd = -1;
int pid_fd_ = -1;
std::string pid_filename_;
std::string config_filename_;
std::string log_config_file_;

2
cpp/src/server/grpc_impl/GrpcRequestTask.cpp
View File

@ -772,7 +772,7 @@ CmdTask::OnExecute() {
if (cmd_ == "version") {
result_ = MILVUS_VERSION;
} else if (cmd_ == "tasktable") {
result_ = engine::ResMgrInst::GetInstance()->DumpTaskTables();
result_ = scheduler::ResMgrInst::GetInstance()->DumpTaskTables();
}
else {
result_ = "OK";

1
cpp/src/utils/Error.h
View File

@ -87,6 +87,7 @@ constexpr ErrorCode DB_NOT_FOUND = ToDbErrorCode(3);
constexpr ErrorCode DB_ALREADY_EXIST = ToDbErrorCode(4);
constexpr ErrorCode DB_INVALID_PATH = ToDbErrorCode(5);
constexpr ErrorCode DB_INCOMPATIB_META = ToDbErrorCode(6);
constexpr ErrorCode DB_INVALID_META_URI = ToDbErrorCode(7);
//knowhere error code
constexpr ErrorCode KNOWHERE_ERROR = ToKnowhereErrorCode(1);

2
cpp/src/utils/Exception.h
View File

@ -28,7 +28,7 @@ namespace milvus {
class Exception : public std::exception {
public:
Exception(ErrorCode code, const std::string& message)
: code_(code_),
: code_(code),
message_(message) {
}

44
cpp/src/utils/ThreadPool.h
View File

@ -45,17 +45,17 @@ public:
private:
// need to keep track of threads so we can join them
std::vector<std::thread> workers;
std::vector<std::thread> workers_;
// the task queue
std::queue<std::function<void()> > tasks;
std::queue<std::function<void()> > tasks_;
size_t max_queue_size;
size_t max_queue_size_;
// synchronization
std::mutex queue_mutex;
std::mutex queue_mutex_;
std::condition_variable condition;
std::condition_variable condition_;
bool stop;
};
@ -63,23 +63,23 @@ private:
// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads, size_t queue_size)
: max_queue_size(queue_size), stop(false) {
: max_queue_size_(queue_size), stop(false) {
for (size_t i = 0; i < threads; ++i)
workers.emplace_back(
workers_.emplace_back(
[this] {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(this->queue_mutex);
this->condition.wait(lock,
[this] { return this->stop || !this->tasks.empty(); });
if (this->stop && this->tasks.empty())
std::unique_lock<std::mutex> lock(this->queue_mutex_);
this->condition_.wait(lock,
[this] { return this->stop || !this->tasks_.empty(); });
if (this->stop && this->tasks_.empty())
return;
task = std::move(this->tasks.front());
this->tasks.pop();
task = std::move(this->tasks_.front());
this->tasks_.pop();
}
this->condition.notify_all();
this->condition_.notify_all();
task();
}
@ -99,27 +99,27 @@ auto ThreadPool::enqueue(F &&f, Args &&... args)
std::future<return_type> res = task->get_future();
{
std::unique_lock<std::mutex> lock(queue_mutex);
this->condition.wait(lock,
[this] { return this->tasks.size() < max_queue_size; });
std::unique_lock<std::mutex> lock(queue_mutex_);
this->condition_.wait(lock,
[this] { return this->tasks_.size() < max_queue_size_; });
// don't allow enqueueing after stopping the pool
if (stop)
throw std::runtime_error("enqueue on stopped ThreadPool");
tasks.emplace([task]() { (*task)(); });
tasks_.emplace([task]() { (*task)(); });
}
condition.notify_all();
condition_.notify_all();
return res;
}
// the destructor joins all threads
inline ThreadPool::~ThreadPool() {
{
std::unique_lock<std::mutex> lock(queue_mutex);
std::unique_lock<std::mutex> lock(queue_mutex_);
stop = true;
}
condition.notify_all();
for (std::thread &worker: workers)
condition_.notify_all();
for (std::thread &worker: workers_)
worker.join();
}

14
cpp/src/wrapper/KnowhereResource.cpp
View File

@ -28,7 +28,8 @@ namespace engine {
constexpr int64_t M_BYTE = 1024 * 1024;
ErrorCode KnowhereResource::Initialize() {
Status
KnowhereResource::Initialize() {
struct GpuResourceSetting {
int64_t pinned_memory = 300*M_BYTE;
int64_t temp_memory = 300*M_BYTE;
@ -43,14 +44,14 @@ ErrorCode KnowhereResource::Initialize() {
int32_t build_index_gpu;
s = config.GetDBConfigBuildIndexGPU(build_index_gpu);
if (!s.ok()) return s.code();
if (!s.ok()) return s;
gpu_resources.insert(std::make_pair(build_index_gpu, GpuResourceSetting()));
//get search gpu resource
std::vector<std::string> pool;
s = config.GetResourceConfigPool(pool);
if (!s.ok()) return s.code();
if (!s.ok()) return s;
std::set<uint64_t> gpu_ids;
for (auto &resource : pool) {
@ -70,12 +71,13 @@ ErrorCode KnowhereResource::Initialize() {
iter->second.resource_num);
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
ErrorCode KnowhereResource::Finalize() {
Status
KnowhereResource::Finalize() {
knowhere::FaissGpuResourceMgr::GetInstance().Free(); // free gpu resource.
return KNOWHERE_SUCCESS;
return Status::OK();
}
}

9
cpp/src/wrapper/KnowhereResource.h
View File

@ -18,7 +18,7 @@
#pragma once
#include "utils/Error.h"
#include "utils/Status.h"
namespace zilliz {
namespace milvus {
@ -26,8 +26,11 @@ namespace engine {
class KnowhereResource {
public:
static ErrorCode Initialize();
static ErrorCode Finalize();
static Status
Initialize();
static Status
Finalize();
};

101
cpp/src/wrapper/vec_impl.cpp
View File

@ -21,7 +21,6 @@
#include "knowhere/index/vector_index/IndexGPUIVF.h"
#include "knowhere/common/Exception.h"
#include "knowhere/index/vector_index/helpers/Cloner.h"
#include "vec_impl.h"
#include "data_transfer.h"
@ -32,7 +31,8 @@ namespace engine {
using namespace zilliz::knowhere;
ErrorCode VecIndexImpl::BuildAll(const long &nb,
Status
VecIndexImpl::BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
@ -49,36 +49,38 @@ ErrorCode VecIndexImpl::BuildAll(const long &nb,
index_->Add(dataset, cfg);
} catch (KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (jsoncons::json_exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_INVALID_ARGUMENT;
return Status(KNOWHERE_INVALID_ARGUMENT, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
ErrorCode VecIndexImpl::Add(const long &nb, const float *xb, const long *ids, const Config &cfg) {
Status
VecIndexImpl::Add(const long &nb, const float *xb, const long *ids, const Config &cfg) {
try {
auto dataset = GenDatasetWithIds(nb, dim, xb, ids);
index_->Add(dataset, cfg);
} catch (KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (jsoncons::json_exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_INVALID_ARGUMENT;
return Status(KNOWHERE_INVALID_ARGUMENT, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
ErrorCode VecIndexImpl::Search(const long &nq, const float *xq, float *dist, long *ids, const Config &cfg) {
Status
VecIndexImpl::Search(const long &nq, const float *xq, float *dist, long *ids, const Config &cfg) {
try {
auto k = cfg["k"].as<int>();
auto dataset = GenDataset(nq, dim, xq);
@ -117,41 +119,47 @@ ErrorCode VecIndexImpl::Search(const long &nq, const float *xq, float *dist, lon
} catch (KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (jsoncons::json_exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_INVALID_ARGUMENT;
return Status(KNOWHERE_INVALID_ARGUMENT, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
zilliz::knowhere::BinarySet VecIndexImpl::Serialize() {
zilliz::knowhere::BinarySet
VecIndexImpl::Serialize() {
type = ConvertToCpuIndexType(type);
return index_->Serialize();
}
ErrorCode VecIndexImpl::Load(const zilliz::knowhere::BinarySet &index_binary) {
Status
VecIndexImpl::Load(const zilliz::knowhere::BinarySet &index_binary) {
index_->Load(index_binary);
dim = Dimension();
return KNOWHERE_SUCCESS;
return Status::OK();
}
int64_t VecIndexImpl::Dimension() {
int64_t
VecIndexImpl::Dimension() {
return index_->Dimension();
}
int64_t VecIndexImpl::Count() {
int64_t
VecIndexImpl::Count() {
return index_->Count();
}
IndexType VecIndexImpl::GetType() {
IndexType
VecIndexImpl::GetType() {
return type;
}
VecIndexPtr VecIndexImpl::CopyToGpu(const int64_t &device_id, const Config &cfg) {
VecIndexPtr
VecIndexImpl::CopyToGpu(const int64_t &device_id, const Config &cfg) {
// TODO(linxj): exception handle
auto gpu_index = zilliz::knowhere::cloner::CopyCpuToGpu(index_, device_id, cfg);
auto new_index = std::make_shared<VecIndexImpl>(gpu_index, ConvertToGpuIndexType(type));
@ -159,7 +167,8 @@ VecIndexPtr VecIndexImpl::CopyToGpu(const int64_t &device_id, const Config &cfg)
return new_index;
}
VecIndexPtr VecIndexImpl::CopyToCpu(const Config &cfg) {
VecIndexPtr
VecIndexImpl::CopyToCpu(const Config &cfg) {
// TODO(linxj): exception handle
auto cpu_index = zilliz::knowhere::cloner::CopyGpuToCpu(index_, cfg);
auto new_index = std::make_shared<VecIndexImpl>(cpu_index, ConvertToCpuIndexType(type));
@ -167,14 +176,16 @@ VecIndexPtr VecIndexImpl::CopyToCpu(const Config &cfg) {
return new_index;
}
VecIndexPtr VecIndexImpl::Clone() {
VecIndexPtr
VecIndexImpl::Clone() {
// TODO(linxj): exception handle
auto clone_index = std::make_shared<VecIndexImpl>(index_->Clone(), type);
clone_index->dim = dim;
return clone_index;
}
int64_t VecIndexImpl::GetDeviceId() {
int64_t
VecIndexImpl::GetDeviceId() {
if (auto device_idx = std::dynamic_pointer_cast<GPUIndex>(index_)) {
return device_idx->GetGpuDevice();
}
@ -182,17 +193,20 @@ int64_t VecIndexImpl::GetDeviceId() {
return -1; // -1 == cpu
}
float *BFIndex::GetRawVectors() {
float *
BFIndex::GetRawVectors() {
auto raw_index = std::dynamic_pointer_cast<IDMAP>(index_);
if (raw_index) { return raw_index->GetRawVectors(); }
return nullptr;
}
int64_t *BFIndex::GetRawIds() {
int64_t *
BFIndex::GetRawIds() {
return std::static_pointer_cast<IDMAP>(index_)->GetRawIds();
}
ErrorCode BFIndex::Build(const Config &cfg) {
ErrorCode
BFIndex::Build(const Config &cfg) {
try {
dim = cfg["dim"].as<int>();
std::static_pointer_cast<IDMAP>(index_)->Train(cfg);
@ -209,7 +223,8 @@ ErrorCode BFIndex::Build(const Config &cfg) {
return KNOWHERE_SUCCESS;
}
ErrorCode BFIndex::BuildAll(const long &nb,
Status
BFIndex::BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
@ -223,19 +238,20 @@ ErrorCode BFIndex::BuildAll(const long &nb,
index_->Add(dataset, cfg);
} catch (KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (jsoncons::json_exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_INVALID_ARGUMENT;
return Status(KNOWHERE_INVALID_ARGUMENT, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
// TODO(linxj): add lock here.
ErrorCode IVFMixIndex::BuildAll(const long &nb,
Status
IVFMixIndex::BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
@ -257,26 +273,27 @@ ErrorCode IVFMixIndex::BuildAll(const long &nb,
type = ConvertToCpuIndexType(type);
} else {
WRAPPER_LOG_ERROR << "Build IVFMIXIndex Failed";
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, "Build IVFMIXIndex Failed");
}
} catch (KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (jsoncons::json_exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_INVALID_ARGUMENT;
return Status(KNOWHERE_INVALID_ARGUMENT, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
ErrorCode IVFMixIndex::Load(const zilliz::knowhere::BinarySet &index_binary) {
Status
IVFMixIndex::Load(const zilliz::knowhere::BinarySet &index_binary) {
//index_ = std::make_shared<IVF>();
index_->Load(index_binary);
dim = Dimension();
return KNOWHERE_SUCCESS;
return Status::OK();
}
}

74
cpp/src/wrapper/vec_impl.h
View File

@ -19,7 +19,6 @@
#pragma once
#include "knowhere/index/vector_index/VectorIndex.h"
#include "vec_index.h"
@ -31,27 +30,53 @@ class VecIndexImpl : public VecIndex {
public:
explicit VecIndexImpl(std::shared_ptr<zilliz::knowhere::VectorIndex> index, const IndexType &type)
: index_(std::move(index)), type(type) {};
ErrorCode BuildAll(const long &nb,
Status
BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
const long &nt,
const float *xt) override;
VecIndexPtr CopyToGpu(const int64_t &device_id, const Config &cfg) override;
VecIndexPtr CopyToCpu(const Config &cfg) override;
IndexType GetType() override;
int64_t Dimension() override;
int64_t Count() override;
ErrorCode Add(const long &nb, const float *xb, const long *ids, const Config &cfg) override;
zilliz::knowhere::BinarySet Serialize() override;
ErrorCode Load(const zilliz::knowhere::BinarySet &index_binary) override;
VecIndexPtr Clone() override;
int64_t GetDeviceId() override;
ErrorCode Search(const long &nq, const float *xq, float *dist, long *ids, const Config &cfg) override;
VecIndexPtr
CopyToGpu(const int64_t &device_id, const Config &cfg) override;
VecIndexPtr
CopyToCpu(const Config &cfg) override;
IndexType
GetType() override;
int64_t
Dimension() override;
int64_t
Count() override;
Status
Add(const long &nb, const float *xb, const long *ids, const Config &cfg) override;
zilliz::knowhere::BinarySet
Serialize() override;
Status
Load(const zilliz::knowhere::BinarySet &index_binary) override;
VecIndexPtr
Clone() override;
int64_t
GetDeviceId() override;
Status
Search(const long &nq, const float *xq, float *dist, long *ids, const Config &cfg) override;
protected:
int64_t dim = 0;
IndexType type = IndexType::INVALID;
std::shared_ptr<zilliz::knowhere::VectorIndex> index_ = nullptr;
};
@ -60,28 +85,39 @@ class IVFMixIndex : public VecIndexImpl {
explicit IVFMixIndex(std::shared_ptr<zilliz::knowhere::VectorIndex> index, const IndexType &type)
: VecIndexImpl(std::move(index), type) {};
ErrorCode BuildAll(const long &nb,
Status
BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
const long &nt,
const float *xt) override;
ErrorCode Load(const zilliz::knowhere::BinarySet &index_binary) override;
Status
Load(const zilliz::knowhere::BinarySet &index_binary) override;
};
class BFIndex : public VecIndexImpl {
public:
explicit BFIndex(std::shared_ptr<zilliz::knowhere::VectorIndex> index) : VecIndexImpl(std::move(index),
IndexType::FAISS_IDMAP) {};
ErrorCode Build(const Config& cfg);
float *GetRawVectors();
ErrorCode BuildAll(const long &nb,
ErrorCode
Build(const Config &cfg);
float *
GetRawVectors();
Status
BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
const long &nt,
const float *xt) override;
int64_t *GetRawIds();
int64_t *
GetRawIds();
};
}

71
cpp/src/wrapper/vec_index.cpp
View File

@ -25,7 +25,6 @@
#include "knowhere/index/vector_index/IndexKDT.h"
#include "knowhere/index/vector_index/IndexNSG.h"
#include "knowhere/common/Exception.h"
#include "vec_index.h"
#include "vec_impl.h"
#include "utils/Log.h"
@ -39,23 +38,19 @@ namespace engine {
static constexpr float TYPICAL_COUNT = 1000000.0;
struct FileIOWriter {
std::fstream fs;
std::string name;
FileIOWriter(const std::string &fname);
~FileIOWriter();
size_t operator()(void *ptr, size_t size);
};
struct FileIOReader {
std::fstream fs;
std::string name;
FileIOReader(const std::string &fname);
~FileIOReader();
size_t operator()(void *ptr, size_t size);
size_t operator()(void *ptr, size_t size, size_t pos);
size_t
operator()(void *ptr, size_t size);
size_t
operator()(void *ptr, size_t size, size_t pos);
};
FileIOReader::FileIOReader(const std::string &fname) {
@ -67,14 +62,27 @@ FileIOReader::~FileIOReader() {
fs.close();
}
size_t FileIOReader::operator()(void *ptr, size_t size) {
size_t
FileIOReader::operator()(void *ptr, size_t size) {
fs.read(reinterpret_cast<char *>(ptr), size);
}
size_t FileIOReader::operator()(void *ptr, size_t size, size_t pos) {
size_t
FileIOReader::operator()(void *ptr, size_t size, size_t pos) {
return 0;
}
struct FileIOWriter {
std::fstream fs;
std::string name;
FileIOWriter(const std::string &fname);
~FileIOWriter();
size_t operator()(void *ptr, size_t size);
};
FileIOWriter::FileIOWriter(const std::string &fname) {
name = fname;
fs = std::fstream(name, std::ios::out | std::ios::binary);
@ -84,12 +92,14 @@ FileIOWriter::~FileIOWriter() {
fs.close();
}
size_t FileIOWriter::operator()(void *ptr, size_t size) {
size_t
FileIOWriter::operator()(void *ptr, size_t size) {
fs.write(reinterpret_cast<char *>(ptr), size);
}
VecIndexPtr GetVecIndexFactory(const IndexType &type, const Config &cfg) {
VecIndexPtr
GetVecIndexFactory(const IndexType &type, const Config &cfg) {
std::shared_ptr<zilliz::knowhere::VectorIndex> index;
auto gpu_device = cfg.get_with_default("gpu_id", 0);
switch (type) {
@ -145,13 +155,15 @@ VecIndexPtr GetVecIndexFactory(const IndexType &type, const Config &cfg) {
return std::make_shared<VecIndexImpl>(index, type);
}
VecIndexPtr LoadVecIndex(const IndexType &index_type, const zilliz::knowhere::BinarySet &index_binary) {
VecIndexPtr
LoadVecIndex(const IndexType &index_type, const zilliz::knowhere::BinarySet &index_binary) {
auto index = GetVecIndexFactory(index_type);
index->Load(index_binary);
return index;
}
VecIndexPtr read_index(const std::string &location) {
VecIndexPtr
read_index(const std::string &location) {
knowhere::BinarySet load_data_list;
FileIOReader reader(location);
reader.fs.seekg(0, reader.fs.end);
@ -195,7 +207,8 @@ VecIndexPtr read_index(const std::string &location) {
return LoadVecIndex(current_type, load_data_list);
}
ErrorCode write_index(VecIndexPtr index, const std::string &location) {
Status
write_index(VecIndexPtr index, const std::string &location) {
try {
auto binaryset = index->Serialize();
auto index_type = index->GetType();
@ -215,23 +228,24 @@ ErrorCode write_index(VecIndexPtr index, const std::string &location) {
}
} catch (knowhere::KnowhereException &e) {
WRAPPER_LOG_ERROR << e.what();
return KNOWHERE_UNEXPECTED_ERROR;
return Status(KNOWHERE_UNEXPECTED_ERROR, e.what());
} catch (std::exception &e) {
WRAPPER_LOG_ERROR << e.what();
std::string estring(e.what());
if (estring.find("No space left on device") != estring.npos) {
WRAPPER_LOG_ERROR << "No space left on the device";
return KNOWHERE_NO_SPACE;
return Status(KNOWHERE_NO_SPACE, "No space left on the device");
} else {
return KNOWHERE_ERROR;
return Status(KNOWHERE_ERROR, e.what());
}
}
return KNOWHERE_SUCCESS;
return Status::OK();
}
// TODO(linxj): redo here.
void AutoGenParams(const IndexType &type, const long &size, zilliz::knowhere::Config &cfg) {
void
AutoGenParams(const IndexType &type, const long &size, zilliz::knowhere::Config &cfg) {
auto nlist = cfg.get_with_default("nlist", 0);
if (size <= TYPICAL_COUNT / 16384 + 1) {
//handle less row count, avoid nlist set to 0
@ -270,7 +284,8 @@ void AutoGenParams(const IndexType &type, const long &size, zilliz::knowhere::Co
#define GPU_MAX_NRPOBE 1024
#endif
void ParameterValidation(const IndexType &type, Config &cfg) {
void
ParameterValidation(const IndexType &type, Config &cfg) {
switch (type) {
case IndexType::FAISS_IVFSQ8_GPU:
case IndexType::FAISS_IVFFLAT_GPU:
@ -291,7 +306,8 @@ void ParameterValidation(const IndexType &type, Config &cfg) {
}
}
IndexType ConvertToCpuIndexType(const IndexType &type) {
IndexType
ConvertToCpuIndexType(const IndexType &type) {
// TODO(linxj): add IDMAP
switch (type) {
case IndexType::FAISS_IVFFLAT_GPU:
@ -308,7 +324,8 @@ IndexType ConvertToCpuIndexType(const IndexType &type) {
}
}
IndexType ConvertToGpuIndexType(const IndexType &type) {
IndexType
ConvertToGpuIndexType(const IndexType &type) {
switch (type) {
case IndexType::FAISS_IVFFLAT_MIX:
case IndexType::FAISS_IVFFLAT_CPU: {

65
cpp/src/wrapper/vec_index.h
View File

@ -21,8 +21,7 @@
#include <string>
#include <memory>
#include "utils/Error.h"
#include "utils/Status.h"
#include "knowhere/common/Config.h"
#include "knowhere/common/BinarySet.h"
@ -50,62 +49,84 @@ enum class IndexType {
};
class VecIndex;
using VecIndexPtr = std::shared_ptr<VecIndex>;
class VecIndex {
public:
virtual ErrorCode BuildAll(const long &nb,
virtual Status
BuildAll(const long &nb,
const float *xb,
const long *ids,
const Config &cfg,
const long &nt = 0,
const float *xt = nullptr) = 0;
virtual ErrorCode Add(const long &nb,
virtual Status
Add(const long &nb,
const float *xb,
const long *ids,
const Config &cfg = Config()) = 0;
virtual ErrorCode Search(const long &nq,
virtual Status
Search(const long &nq,
const float *xq,
float *dist,
long *ids,
const Config &cfg = Config()) = 0;
virtual VecIndexPtr CopyToGpu(const int64_t &device_id,
virtual VecIndexPtr
CopyToGpu(const int64_t &device_id,
const Config &cfg = Config()) = 0;
virtual VecIndexPtr CopyToCpu(const Config &cfg = Config()) = 0;
virtual VecIndexPtr
CopyToCpu(const Config &cfg = Config()) = 0;
virtual VecIndexPtr Clone() = 0;
virtual VecIndexPtr
Clone() = 0;
virtual int64_t GetDeviceId() = 0;
virtual int64_t
GetDeviceId() = 0;
virtual IndexType GetType() = 0;
virtual IndexType
GetType() = 0;
virtual int64_t Dimension() = 0;
virtual int64_t
Dimension() = 0;
virtual int64_t Count() = 0;
virtual int64_t
Count() = 0;
virtual zilliz::knowhere::BinarySet Serialize() = 0;
virtual zilliz::knowhere::BinarySet
Serialize() = 0;
virtual ErrorCode Load(const zilliz::knowhere::BinarySet &index_binary) = 0;
virtual Status
Load(const zilliz::knowhere::BinarySet &index_binary) = 0;
};
extern ErrorCode write_index(VecIndexPtr index, const std::string &location);
extern Status
write_index(VecIndexPtr index, const std::string &location);
extern VecIndexPtr read_index(const std::string &location);
extern VecIndexPtr
read_index(const std::string &location);
extern VecIndexPtr GetVecIndexFactory(const IndexType &type, const Config &cfg = Config());
extern VecIndexPtr
GetVecIndexFactory(const IndexType &type, const Config &cfg = Config());
extern VecIndexPtr LoadVecIndex(const IndexType &index_type, const zilliz::knowhere::BinarySet &index_binary);
extern VecIndexPtr
LoadVecIndex(const IndexType &index_type, const zilliz::knowhere::BinarySet &index_binary);
extern void AutoGenParams(const IndexType &type, const long &size, Config &cfg);
extern void
AutoGenParams(const IndexType &type, const long &size, Config &cfg);
extern void ParameterValidation(const IndexType &type, Config &cfg);
extern void
ParameterValidation(const IndexType &type, Config &cfg);
extern IndexType ConvertToCpuIndexType(const IndexType &type);
extern IndexType ConvertToGpuIndexType(const IndexType &type);
extern IndexType
ConvertToCpuIndexType(const IndexType &type);
extern IndexType
ConvertToGpuIndexType(const IndexType &type);
}
}

8
cpp/unittest/db/meta_tests.cpp
View File

@ -117,11 +117,11 @@ TEST_F(MetaTest, TABLE_FILE_TEST) {
TEST_F(MetaTest, ARCHIVE_TEST_DAYS) {
srand(time(0));
DBMetaOptions options;
options.path = "/tmp/milvus_test";
options.path_ = "/tmp/milvus_test";
int days_num = rand() % 100;
std::stringstream ss;
ss << "days:" << days_num;
options.archive_conf = ArchiveConf("delete", ss.str());
options.archive_conf_ = ArchiveConf("delete", ss.str());
meta::SqliteMetaImpl impl(options);
auto table_id = "meta_test_table";
@ -168,8 +168,8 @@ TEST_F(MetaTest, ARCHIVE_TEST_DAYS) {
TEST_F(MetaTest, ARCHIVE_TEST_DISK) {
DBMetaOptions options;
options.path = "/tmp/milvus_test";
options.archive_conf = ArchiveConf("delete", "disk:11");
options.path_ = "/tmp/milvus_test";
options.archive_conf_ = ArchiveConf("delete", "disk:11");
meta::SqliteMetaImpl impl(options);
auto table_id = "meta_test_group";

12
cpp/unittest/db/misc_test.cpp
View File

@ -77,7 +77,7 @@ TEST(DBMiscTest, OPTIONS_TEST) {
TEST(DBMiscTest, META_TEST) {
engine::DBMetaOptions options;
options.path = "/tmp/milvus_test";
options.path_ = "/tmp/milvus_test";
engine::meta::SqliteMetaImpl impl(options);
time_t tt;
@ -89,15 +89,15 @@ TEST(DBMiscTest, META_TEST) {
TEST(DBMiscTest, UTILS_TEST) {
engine::DBMetaOptions options;
options.path = "/tmp/milvus_test/main";
options.slave_paths.push_back("/tmp/milvus_test/slave_1");
options.slave_paths.push_back("/tmp/milvus_test/slave_2");
options.path_ = "/tmp/milvus_test/main";
options.slave_paths_.push_back("/tmp/milvus_test/slave_1");
options.slave_paths_.push_back("/tmp/milvus_test/slave_2");
const std::string TABLE_NAME = "test_tbl";
auto status = engine::utils::CreateTablePath(options, TABLE_NAME);
ASSERT_TRUE(status.ok());
ASSERT_TRUE(boost::filesystem::exists(options.path));
for(auto& path : options.slave_paths) {
ASSERT_TRUE(boost::filesystem::exists(options.path_));
for(auto& path : options.slave_paths_) {
ASSERT_TRUE(boost::filesystem::exists(path));
}

8
cpp/unittest/db/mysql_meta_test.cpp
View File

@ -121,12 +121,12 @@ TEST_F(MySqlMetaTest, TABLE_FILE_TEST) {
TEST_F(MySqlMetaTest, ARCHIVE_TEST_DAYS) {
srand(time(0));
DBMetaOptions options = GetOptions().meta;
DBMetaOptions options = GetOptions().meta_;
int days_num = rand() % 100;
std::stringstream ss;
ss << "days:" << days_num;
options.archive_conf = ArchiveConf("delete", ss.str());
options.archive_conf_ = ArchiveConf("delete", ss.str());
int mode = DBOptions::MODE::SINGLE;
meta::MySQLMetaImpl impl(options, mode);
@ -184,9 +184,9 @@ TEST_F(MySqlMetaTest, ARCHIVE_TEST_DAYS) {
}
TEST_F(MySqlMetaTest, ARCHIVE_TEST_DISK) {
DBMetaOptions options = GetOptions().meta;
DBMetaOptions options = GetOptions().meta_;
options.archive_conf = ArchiveConf("delete", "disk:11");
options.archive_conf_ = ArchiveConf("delete", "disk:11");
int mode = DBOptions::MODE::SINGLE;
auto impl = meta::MySQLMetaImpl(options, mode);
auto table_id = "meta_test_group";

Some files were not shown because too many files have changed in this diff Show More