niklas/cyclonedds
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix read/take/return_loan edge cases

Browse source 
* read/take failed to restore the null pointer in the first entry of the
  sample pointer array it gets passed, in the case no "loan" had been
  allocated yet and it returned an empty set.  The consequence is that
  on a subsequence read it will reuse the address without marking at as
  in use, so that a *second* read using with a null pointer in that
  first entry will overwrite the first result.  (Introduced by
  d16264fd82.)

* return_loan failed to free all memory if its argument wasn't actually
  a loan.  There are many good arguments why the read/take/return_loan
  interface is messed up, but in the context of the existing interface
  this is a perfectly reasonable case: there is at most one "loan" for
  each reader, but one can keep calling read/take and return_loan as if
  there's an infinite number of "loans".  It's just that the first gets
  cached and the others don't.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2020-03-24 18:51:29 +01:00 • committed by eboasson
parent d089ce946c
commit 5d53e74029
4 changed files with 363 additions and 154 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/core/ddsc/src/dds_read.c
View file

@ -94,7 +94,6 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
rd->m_loan = buf[0];
rd->m_loan_size = maxs;
}
nodata_cleanups = NC_RESET_BUF;
}
else
{
@ -103,7 +102,7 @@ static dds_return_t dds_read_impl (bool take, dds_entity_t reader_or_condition,
rd->m_loan_size = maxs;
}
rd->m_loan_out = true;
nodata_cleanups |= NC_CLEAR_LOAN_OUT;
nodata_cleanups = NC_RESET_BUF | NC_CLEAR_LOAN_OUT;
}
ddsrt_mutex_unlock (&rd->m_entity.m_mutex);
}
@ -457,12 +456,11 @@ dds_return_t dds_take_next_wl (dds_entity_t reader, void **buf, dds_sample_info_
dds_return_t dds_return_loan (dds_entity_t reader_or_condition, void **buf, int32_t bufsz)
{
const struct ddsi_sertopic *st;
dds_reader *rd;
dds_entity *entity;
dds_return_t ret = DDS_RETCODE_OK;
dds_return_t ret;
if (buf == NULL || (*buf == NULL && bufsz > 0))
if (buf == NULL || (buf[0] == NULL && bufsz > 0) || (buf[0] != NULL && bufsz <= 0))
return DDS_RETCODE_BAD_PARAMETER;
if ((ret = dds_entity_pin (reader_or_condition, &entity)) < 0) {
@ -476,16 +474,46 @@ dds_return_t dds_return_loan (dds_entity_t reader_or_condition, void **buf, int3
rd = (dds_reader *) entity->m_parent;
}
st = rd->m_topic->m_stopic;
for (int32_t i = 0; i < bufsz; i++)
ddsi_sertopic_free_sample (st, buf[i], DDS_FREE_CONTENTS);
/* If possible return loan buffer to reader */
ddsrt_mutex_lock (&rd->m_entity.m_mutex);
if (rd->m_loan != 0 && (buf[0] == rd->m_loan))
if (bufsz <= 0)
{
rd->m_loan_out = false;
/* No data whatsoever, or an invocation following a failed read/take call. Read/take
already take care of restoring the state prior to their invocation if they return
no data. Return late so invalid handles can be detected. */
dds_entity_unpin (entity);
return DDS_RETCODE_OK;
}
assert (buf[0] != NULL);
const struct ddsi_sertopic *st = rd->m_topic->m_stopic;
/* The potentially time consuming part of what happens here (freeing samples)
can safely be done without holding the reader lock, but that particular
lock is not used during insertion of data & triggering waitsets (that's
the observer_lock), so holding it for a bit longer in return for simpler
code is a fair trade-off. */
ddsrt_mutex_lock (&rd->m_entity.m_mutex);
if (buf[0] != rd->m_loan)
{
/* Not so much a loan as a buffer allocated by the middleware on behalf of the
application. So it really is no more than a sophisticated variant of "free". */
ddsi_sertopic_free_samples (st, buf, (size_t) bufsz, DDS_FREE_ALL);
buf[0] = NULL;
}
else if (!rd->m_loan_out)
{
/* Trying to return a loan that has been returned already */
ddsrt_mutex_unlock (&rd->m_entity.m_mutex);
dds_entity_unpin (entity);
return DDS_RETCODE_PRECONDITION_NOT_MET;
}
else
{
/* Free only the memory referenced from the samples, not the samples themselves.
Zero them to guarantee the absence of dangling pointers that might cause
trouble on a following operation. FIXME: there's got to be a better way */
ddsi_sertopic_free_samples (st, buf, (size_t) bufsz, DDS_FREE_CONTENTS);
ddsi_sertopic_zero_samples (st, rd->m_loan, rd->m_loan_size);
rd->m_loan_out = false;
buf[0] = NULL;
}
ddsrt_mutex_unlock (&rd->m_entity.m_mutex);

2
src/core/ddsc/tests/CMakeLists.txt
View file

@ -30,6 +30,7 @@ set(ddsc_test_sources
"instance_get_key.c"
"listener.c"
"liveliness.c"
"loan.c"
"multi_sertopic.c"
"participant.c"
"publisher.c"
@ -41,7 +42,6 @@ set(ddsc_test_sources
"reader_iterator.c"
"read_instance.c"
"register.c"
"return_loan.c"
"subscriber.c"
"take_instance.c"
"time.c"

321
src/core/ddsc/tests/loan.c Normal file
View file

@ -0,0 +1,321 @@
/*
* Copyright(c) 2006 to 2020 ADLINK Technology Limited and others
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v. 2.0 which is available at
* http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
* v. 1.0 which is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
*/
#include <stdio.h>
#include "dds/dds.h"
#include "test_common.h"
static dds_entity_t participant, topic, reader, writer, read_condition, read_condition_unread;
static void create_entities (void)
{
char topicname[100];
struct dds_qos *qos;
create_unique_topic_name ("ddsc_return_loan_test", topicname, sizeof topicname);
participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
CU_ASSERT_FATAL (participant > 0);
qos = dds_create_qos ();
dds_qset_reliability (qos, DDS_RELIABILITY_RELIABLE, 0);
dds_qset_history (qos, DDS_HISTORY_KEEP_ALL, 1);
topic = dds_create_topic (participant, &RoundTripModule_DataType_desc, topicname, qos, NULL);
CU_ASSERT_FATAL (topic > 0);
dds_delete_qos (qos);
writer = dds_create_writer (participant, topic, NULL, NULL);
CU_ASSERT_FATAL (writer > 0);
reader = dds_create_reader (participant, topic, NULL, NULL);
CU_ASSERT_FATAL (reader > 0);
read_condition = dds_create_readcondition (reader, DDS_ANY_STATE);
CU_ASSERT_FATAL (read_condition > 0);
read_condition_unread = dds_create_readcondition (reader, DDS_ANY_INSTANCE_STATE | DDS_ANY_VIEW_STATE | DDS_NOT_READ_SAMPLE_STATE);
CU_ASSERT_FATAL (read_condition > 0);
}
static void delete_entities (void)
{
dds_return_t result;
result = dds_delete (participant);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
}
CU_Test (ddsc_loan, bad_params, .init = create_entities, .fini = delete_entities)
{
dds_return_t result;
/* buf = NULL */
result = dds_return_loan (reader, NULL, -1);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
result = dds_return_loan (reader, NULL, 0);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
result = dds_return_loan (reader, NULL, 1);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
/* buf[0] = NULL, size > 0 */
void *buf = NULL;
result = dds_return_loan (reader, &buf, 1);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
/* buf[0] != NULL, size <= 0 */
char dummy = 0;
buf = &dummy;
result = dds_return_loan (reader, &buf, 0);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
result = dds_return_loan (reader, &buf, -1);
CU_ASSERT (result == DDS_RETCODE_BAD_PARAMETER);
/* not a reader or condition (checking only the ones we have at hand) */
result = dds_return_loan (participant, &buf, 1);
CU_ASSERT (result == DDS_RETCODE_ILLEGAL_OPERATION);
result = dds_return_loan (topic, &buf, 1);
CU_ASSERT (result == DDS_RETCODE_ILLEGAL_OPERATION);
}
CU_Test (ddsc_loan, success, .init = create_entities, .fini = delete_entities)
{
const RoundTripModule_DataType s = {
.payload = {
._length = 1,
._buffer = (uint8_t[]) { 'a' }
}
};
const unsigned char zeros[3 * sizeof (s)] = { 0 };
dds_return_t result;
for (size_t i = 0; i < 3; i++)
{
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
}
/* rely on things like address sanitizer, valgrind for detecting double frees and leaks */
int32_t n;
void *ptrs[3] = { NULL };
void *ptr0copy, *ptr1copy;
dds_sample_info_t si[3];
/* read 1, return: this should cause memory to be allocated for 1 sample only */
n = dds_read (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] != NULL && ptrs[1] == NULL);
ptr0copy = ptrs[0];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
/* return resets buf[0] (so that it picks up the loan the next time) and zeros the data */
CU_ASSERT_FATAL (ptrs[0] == NULL);
CU_ASSERT_FATAL (memcmp (ptr0copy, zeros, sizeof (s)) == 0);
/* read 3, return: should work fine, causes realloc */
n = dds_read (reader, ptrs, si, 3, 3);
CU_ASSERT_FATAL (n == 3);
CU_ASSERT_FATAL (ptrs[0] != NULL && ptrs[1] != NULL && ptrs[2] != NULL);
ptr0copy = ptrs[0];
ptr1copy = ptrs[1];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
CU_ASSERT_FATAL (ptrs[0] == NULL);
CU_ASSERT_FATAL (memcmp (ptr0copy, zeros, 3 * sizeof (s)) == 0);
/* read 1 using loan, expecting to get the same address (no realloc needed), defer return.
Expect ptrs[1] to remain unchanged, although that probably is really an implementation
detail rather than something one might want to rely on */
n = dds_read (read_condition, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == ptr1copy);
/* read 3, letting read allocate */
int32_t n2;
void *ptrs2[3] = { NULL };
n2 = dds_read (read_condition, ptrs2, si, 3, 3);
CU_ASSERT_FATAL (n2 == 3);
CU_ASSERT_FATAL (ptrs2[0] != NULL && ptrs2[1] != NULL && ptrs2[2] != NULL);
CU_ASSERT_FATAL (ptrs2[0] != ptrs[0]);
/* contents of first sample should be the same; the point of comparing them
is that valgrind/address sanitizer will get angry with us if one of them
has been freed; can't use memcmp because the sequence buffers should be
at different addresses */
{
const struct RoundTripModule_DataType *a = ptrs[0];
const struct RoundTripModule_DataType *b = ptrs2[0];
CU_ASSERT_FATAL (a->payload._length == b->payload._length);
CU_ASSERT_FATAL (a->payload._buffer != b->payload._buffer);
CU_ASSERT_FATAL (a->payload._buffer[0] == b->payload._buffer[0]);
}
/* return loan -- to be freed when we delete the reader */
result = dds_return_loan (read_condition, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
CU_ASSERT_FATAL (ptrs[0] == NULL);
/* use "dds_return_loan" to free the second result immediately, there's no
easy way to check this happens short of using a custom sertopic */
ptr0copy = ptrs2[0];
result = dds_return_loan (read_condition, ptrs2, n2);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
CU_ASSERT_FATAL (ptrs2[0] == NULL);
//This should be a use-after-free
//CU_ASSERT_FATAL (memcmp (ptr0copy, zeros, sizeof (s)) == 0);
}
CU_Test (ddsc_loan, take_cleanup, .init = create_entities, .fini = delete_entities)
{
const RoundTripModule_DataType s = {
.payload = {
._length = 1,
._buffer = (uint8_t[]) { 'a' }
}
};
dds_return_t result;
/* rely on things like address sanitizer, valgrind for detecting double frees and leaks */
int32_t n;
void *ptrs[3] = { NULL };
void *ptr0copy;
dds_sample_info_t si[3];
/* take 1 from an empty reader: this should cause memory to be allocated for
1 sample only, be stored as the loan, but not become visisble to the
application */
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 0);
CU_ASSERT_FATAL (ptrs[0] == NULL && ptrs[1] == NULL);
/* take 1 that's present: allocates a loan */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] != NULL && ptrs[1] == NULL);
ptr0copy = ptrs[0];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
/* if it really got handled as a loan, the same address must come out again
(rely on address sanitizer allocating at a different address each time) */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
ptr0copy = ptrs[0];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
/* take that fails (for lack of data in this case) must reuse the loan, but
hand it back and restore the null pointer */
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 0);
CU_ASSERT_FATAL (ptrs[0] == NULL && ptrs[1] == NULL);
/* take that succeeds again must therefore still be using the same address */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
/* take that fails (with the loan still out) must allocate new memory and
free it */
int32_t n2;
void *ptrs2[3] = { NULL };
n2 = dds_take (reader, ptrs2, si, 1, 1);
CU_ASSERT_FATAL (n2 == 0);
CU_ASSERT_FATAL (ptrs2[0] == NULL && ptrs2[1] == NULL);
/* return the loan and the next take should reuse the memory */
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
}
CU_Test (ddsc_loan, read_cleanup, .init = create_entities, .fini = delete_entities)
{
const RoundTripModule_DataType s = {
.payload = {
._length = 1,
._buffer = (uint8_t[]) { 'a' }
}
};
dds_return_t result;
/* rely on things like address sanitizer, valgrind for detecting double frees and leaks */
int32_t n;
void *ptrs[3] = { NULL };
void *ptr0copy;
dds_sample_info_t si[3];
/* read 1 from an empty reader: this should cause memory to be allocated for
1 sample only, be stored as the loan, but not become visisble to the
application */
n = dds_read (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 0);
CU_ASSERT_FATAL (ptrs[0] == NULL && ptrs[1] == NULL);
/* read 1 that's present: allocates a loan */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_take (reader, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] != NULL && ptrs[1] == NULL);
ptr0copy = ptrs[0];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
/* if it really got handled as a loan, the same address must come out again
(rely on address sanitizer allocating at a different address each time) */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_read (read_condition_unread, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
ptr0copy = ptrs[0];
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
/* take that fails (for lack of data in this case) must reuse the loan, but
hand it back and restore the null pointer */
n = dds_read (read_condition_unread, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 0);
CU_ASSERT_FATAL (ptrs[0] == NULL && ptrs[1] == NULL);
/* take that succeeds again must therefore still be using the same address */
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_read (read_condition_unread, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
/* take that fails (with the loan still out) must allocate new memory and
free it */
int32_t n2;
void *ptrs2[3] = { NULL };
n2 = dds_read (read_condition_unread, ptrs2, si, 1, 1);
CU_ASSERT_FATAL (n2 == 0);
CU_ASSERT_FATAL (ptrs2[0] == NULL && ptrs2[1] == NULL);
/* return the loan and the next take should reuse the memory */
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
result = dds_write (writer, &s);
CU_ASSERT_FATAL (result == 0);
n = dds_read (read_condition_unread, ptrs, si, 1, 1);
CU_ASSERT_FATAL (n == 1);
CU_ASSERT_FATAL (ptrs[0] == ptr0copy && ptrs[1] == NULL);
result = dds_return_loan (reader, ptrs, n);
CU_ASSERT_FATAL (result == DDS_RETCODE_OK);
}

140
src/core/ddsc/tests/return_loan.c
View file

@ -1,140 +0,0 @@
/*
* Copyright(c) 2006 to 2018 ADLINK Technology Limited and others
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v. 2.0 which is available at
* http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
* v. 1.0 which is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
*/
#include "dds/dds.h"
#include "RoundTrip.h"
#include <stdio.h>
#include "CUnit/Test.h"
static dds_entity_t participant = 0, topic = 0, reader = 0, read_condition = 0;
static void create_entities(void)
{
participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
CU_ASSERT_FATAL(participant > 0);
topic = dds_create_topic(participant, &RoundTripModule_DataType_desc, "ddsc_reader_return_loan_RoundTrip", NULL, NULL);
CU_ASSERT_FATAL(topic > 0);
reader = dds_create_reader(participant, topic, NULL, NULL);
CU_ASSERT_FATAL(reader > 0);
read_condition = dds_create_readcondition(reader, DDS_ANY_STATE);
CU_ASSERT_FATAL(read_condition > 0);
}
static void delete_entities(void)
{
dds_return_t result;
result = dds_delete(participant);
CU_ASSERT_EQUAL_FATAL(result, DDS_RETCODE_OK);
dds_delete(read_condition);
}
static void** create_loan_buf(size_t sz, bool empty)
{
size_t i;
void **buf = NULL;
buf = dds_alloc(sz * sizeof(*buf));
for (i = 0; i < sz; i++) {
buf[i] = dds_alloc(sizeof(RoundTripModule_DataType));
if (empty) {
memset(buf[i], 0, sizeof(RoundTripModule_DataType));
} else {
RoundTripModule_DataType *s = buf[i];
s->payload._maximum = 0;
s->payload._length = 25;
s->payload._buffer = dds_alloc(25);
memset(s->payload._buffer, 'z', 25);
s->payload._release = true;
}
}
return buf;
}
static void delete_loan_buf(void **buf, size_t sz, bool empty)
{
size_t i;
for (i = 0; i < sz; i++) {
RoundTripModule_DataType *s = buf[i];
if (!empty) {
CU_ASSERT(s->payload._length > 0);
if (s->payload._length > 0) {
/* Freed by a successful dds_return_loan */
dds_free(s->payload._buffer);
}
}
/* dds_return_loan only free's sample contents */
dds_free(s);
}
dds_free(buf);
}
/* Verify DDS_RETCODE_BAD_PARAMETER is returned */
CU_Test(ddsc_reader, return_loan_bad_params, .init = create_entities, .fini = delete_entities)
{
dds_return_t result;
void **buf = NULL;
result = dds_return_loan(reader, NULL, 0);
CU_ASSERT_EQUAL(result, DDS_RETCODE_BAD_PARAMETER);
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable: 6387)
#endif
result = dds_return_loan(reader, buf, 10);
#ifdef _MSC_VER
#pragma warning(pop)
#endif
CU_ASSERT_EQUAL(result, DDS_RETCODE_BAD_PARAMETER);
buf = create_loan_buf(10, false);
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable: 28020)
#endif
result = dds_return_loan(0, buf, 10);
#ifdef _MSC_VER
#pragma warning(pop)
#endif
CU_ASSERT_EQUAL(result, DDS_RETCODE_BAD_PARAMETER);
result = dds_return_loan(participant, buf, 0);
CU_ASSERT_EQUAL(result, DDS_RETCODE_ILLEGAL_OPERATION);
delete_loan_buf(buf, 10, false);
}
/* Verify DDS_RETCODE_OK is returned */
CU_Test(ddsc_reader, return_loan_success, .init = create_entities, .fini = delete_entities)
{
void **buf;
void *buf2 = NULL;
dds_return_t result;
buf = create_loan_buf(10, false);
result = dds_return_loan(reader, buf, 10);
CU_ASSERT_EQUAL(result, DDS_RETCODE_OK);
result = dds_return_loan(reader, &buf2, 0);
CU_ASSERT_EQUAL(result, DDS_RETCODE_OK);
delete_loan_buf(buf, 10, true);
buf = create_loan_buf(10, false);
result = dds_return_loan(read_condition, buf, 10);
CU_ASSERT_EQUAL(result, DDS_RETCODE_OK);
result = dds_return_loan(read_condition, &buf2, 0);
CU_ASSERT_EQUAL(result, DDS_RETCODE_OK);
delete_loan_buf(buf, 10, true);
}