Browse Source

CacheOp delivery for HQ tester

tags/v1.1.0
Jesse Lee 5 years ago
parent
commit
c795879313
5 changed files with 41 additions and 10 deletions
  1. +5
    -1
      mindspore/ccsrc/minddata/dataset/engine/cache/cache_admin_arg.cc
  2. +3
    -0
      mindspore/ccsrc/minddata/dataset/engine/cache/cache_client.cc
  3. +4
    -6
      mindspore/ccsrc/minddata/dataset/engine/cache/cache_pool.cc
  4. +1
    -1
      tests/ut/python/cachetests/cachetest_args.sh
  5. +28
    -2
      tests/ut/python/dataset/test_cache_nomap.py

+ 5
- 1
mindspore/ccsrc/minddata/dataset/engine/cache/cache_admin_arg.cc View File

@@ -17,6 +17,7 @@
#include <sys/types.h> #include <sys/types.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <sys/wait.h> #include <sys/wait.h>
#include <signal.h>
#include <unistd.h> #include <unistd.h>
#include <algorithm> #include <algorithm>
#include <cerrno> #include <cerrno>
@@ -351,7 +352,10 @@ Status CacheAdminArgHandler::RunCommand() {
// have to wait for its complete shutdown because the server will shutdown // have to wait for its complete shutdown because the server will shutdown
// the comm layer as soon as the request is received, and we need to wait // the comm layer as soon as the request is received, and we need to wait
// on the message queue instead. // on the message queue instead.
// The server will remove the queue and we will then wake up.
// The server will remove the queue and we will then wake up. But on the safe
// side, we will also set up an alarm and kill this proocess if we hang on
// the message queue.
alarm(15);
Status dummy_rc; Status dummy_rc;
(void)msg.ReceiveStatus(&dummy_rc); (void)msg.ReceiveStatus(&dummy_rc);
break; break;


+ 3
- 0
mindspore/ccsrc/minddata/dataset/engine/cache/cache_client.cc View File

@@ -230,6 +230,9 @@ Status CacheClient::DestroyCache() {
Status CacheClient::GetStat(CacheServiceStat *stat) { Status CacheClient::GetStat(CacheServiceStat *stat) {
SharedLock lck(&mux_); SharedLock lck(&mux_);
RETURN_UNEXPECTED_IF_NULL(stat); RETURN_UNEXPECTED_IF_NULL(stat);
// GetStat has an external interface, so we have to make sure we have a valid connection id first
CHECK_FAIL_RETURN_UNEXPECTED(server_connection_id_ != 0, "GetStat called but the cache is not in use yet.");

auto rq = std::make_shared<GetStatRequest>(server_connection_id_); auto rq = std::make_shared<GetStatRequest>(server_connection_id_);
RETURN_IF_NOT_OK(PushRequest(rq)); RETURN_IF_NOT_OK(PushRequest(rq));
RETURN_IF_NOT_OK(rq->Wait()); RETURN_IF_NOT_OK(rq->Wait());


+ 4
- 6
mindspore/ccsrc/minddata/dataset/engine/cache/cache_pool.cc View File

@@ -48,12 +48,10 @@ Status CachePool::DoServiceStop() {
} }
sm_.reset(); sm_.reset();


value_allocator alloc(mp_);
for (auto &bl : *tree_) {
if (bl.ptr != nullptr) {
alloc.deallocate(bl.ptr, bl.sz);
}
}
// We used to free the memory allocated from each DataLocator but
// since all of them are coming from NumaMemoryPool and we will
// skip this and release the whole NumaMemoryPool instead. Otherwise
// release each buffer in the DataLocator one by one.


tree_.reset(); tree_.reset();
if (!root_.toString().empty()) { if (!root_.toString().empty()) {


+ 1
- 1
tests/ut/python/cachetests/cachetest_args.sh View File

@@ -175,7 +175,7 @@ num_cpu=$(grep -c processor /proc/cpuinfo)
if [ $num_cpu -lt 100 ]; then if [ $num_cpu -lt 100 ]; then
cmd="${CACHE_ADMIN} --start -w 101" cmd="${CACHE_ADMIN} --start -w 101"
else else
cmd="${CACHE_ADMIN} --start -w ${num_cpu}+1"
cmd="${CACHE_ADMIN} --start -w $(($num_cpu+1))"
fi fi
CacheAdminCmd "${cmd}" 1 CacheAdminCmd "${cmd}" 1
HandleRcExit $? 0 0 HandleRcExit $? 0 0


+ 28
- 2
tests/ut/python/dataset/test_cache_nomap.py View File

@@ -339,7 +339,7 @@ def test_cache_nomap_basic8():
| |
TFReader TFReader
""" """
logger.info("Test cache basic 4")
logger.info("Test cache basic 8")
if "SESSION_ID" in os.environ: if "SESSION_ID" in os.environ:
session_id = int(os.environ['SESSION_ID']) session_id = int(os.environ['SESSION_ID'])
else: else:
@@ -355,7 +355,33 @@ def test_cache_nomap_basic8():


logger.info("Number of data in ds1: {} ".format(num_iter)) logger.info("Number of data in ds1: {} ".format(num_iter))
assert num_iter == 3 assert num_iter == 3
logger.info('test_cache_basic3 Ended.\n')
logger.info('test_cache_basic8 Ended.\n')


@pytest.mark.skipif(os.environ.get('RUN_CACHE_TEST') != 'TRUE', reason="Require to bring up cache server")
def test_cache_nomap_basic9():
"""
Testing the GetStat interface for getting some info from server, but this should fail if the cache is not created
in a pipeline.
"""

logger.info("Test cache nomap basic 9")
if "SESSION_ID" in os.environ:
session_id = int(os.environ['SESSION_ID'])
else:
raise RuntimeError("Testcase requires SESSION_ID environment variable")

some_cache = ds.DatasetCache(session_id=session_id, size=0, spilling=True)

# Contact the server to get the statistics, this should fail because we have not used this cache in any pipeline
# so there will not be any cache to get stats on.
with pytest.raises(RuntimeError) as e:
stat = some_cache.GetStat()
cache_sz = stat.avg_cache_sz
logger.info("Average row cache size: {}".format(cache_sz))
assert "Unexpected error" in str(e.value)

logger.info("test_cache_nomap_basic9 Ended.\n")




@pytest.mark.skipif(os.environ.get('RUN_CACHE_TEST') != 'TRUE', reason="Require to bring up cache server") @pytest.mark.skipif(os.environ.get('RUN_CACHE_TEST') != 'TRUE', reason="Require to bring up cache server")


Loading…
Cancel
Save