Skip to content

P
proview
Commit a41fe692 authored by Claes Sjofors's avatar Claes Sjofors
Browse files
Options

neth messages for circular buffers added

parent 187a55a0
Hide whitespace changes
Inline Side-by-side
Showing with 484 additions and 4 deletions
src/exe/rt_neth/src/rt_neth.c
View file @ a41fe692
......@@ -100,6 +100,8 @@ static void volumesR (qcom_sGet*);
static void volumes7 (qcom_sGet*);
static void serverConnect (qcom_sGet*);
static void fileList (qcom_sGet*);
static void getCircBuffer (qcom_sGet*);
static void updateCircBuffer(qcom_sGet*);
#if 0
static void linkEvent (pwr_tUInt32, net_eEvent);
static void sendIdAck (gdb_sNode*);
......@@ -142,6 +144,11 @@ static char *cMsg[net_eMsg_end] = {
"getGclassR",
"serverConnect",
"fileList",
"fileListR",
"getCircBuffer",
"getCircBufferR",
"updateCircBuffer",
"updateCircBufferR",
"net_eMsg_",
"volumes7"
};
......@@ -178,6 +185,11 @@ static void (*fromApplication[net_eMsg_end])(qcom_sGet *) = {
bugError, /* net_eMsg_GetGclassR will never reach neth */
serverConnect, /* net_eMsg_serverConnect */
fileList, /* net_eMsg_fileList */
bugError, /* net_eMsg_fileListR, will never reach neth. */
getCircBuffer, /* net_eMsg_getCircBuffer */
bugError, /* net_eMsg_getCircBufferR, will never reach neth. */
updateCircBuffer, /* net_eMsg_updateCircBuffer */
bugError, /* net_eMsg_updateCircBufferR, will never reach neth. */
bugError, /* net_eMsg_ */
volumes7
};
......@@ -1554,3 +1566,150 @@ fileList (
net_Reply(&sts, get, &put, 0);
}
static void
getCircBuffer (
qcom_sGet *get
)
{
net_sGetCircBuffer *mp = get->data;
net_sGetCircBufferR *rmp;
gdb_sNode *np;
pwr_tStatus sts;
pwr_tUInt32 size;
qcom_sPut put;
gdh_sCircBuffInfo info;
gdb_ScopeLock {
np = hash_Search(&sts, gdbroot->nid_ht, &mp->hdr.nid);
} gdb_ScopeUnlock;
if (gdbroot->db->log.b.id) {
errh_Info("Sending 'getCircBuffer' to %s (%s)",
np->name, cdh_NodeIdToString(NULL, np->nid, 0, 0));
}
info.circ_aref = mp->circ_aref;
info.resolution = mp->resolution;
info.samples = mp->samples;
info.bufsize = mp->bufsize;
info.bufp = calloc( 1, info.bufsize);
sts = gdh_GetCircBuffInfo( &info, 1);
if ( EVEN(sts))
size = sizeof(*rmp);
else
size = sizeof(*rmp) + info.bufsize;
size = (size + 3) & ~3; /* Size up to nearest multiple of 4. */
rmp = net_Alloc(&sts, &put, size, net_eMsg_getCircBufferR);
if (rmp == NULL) {
errh_Error("Failed to allocate 'getCircBufferR' to %s (%s)",
np->name, cdh_NodeIdToString(NULL, np->nid, 0, 0));
return;
}
rmp->sts = sts;
if ( EVEN(sts)) {
rmp->sts = sts;
rmp->size = 0;
}
else {
rmp->sts = sts;
rmp->circ_aref = info.circ_aref;
rmp->size = info.size;
rmp->bufsize = info.bufsize;
rmp->first_idx = info.first_idx;
rmp->last_idx = info.last_idx;
rmp->offset = info.offset;
memcpy( rmp->buf, info.bufp, info.bufsize);
free( info.bufp);
}
net_Reply(&sts, get, &put, 0);
}
static void
updateCircBuffer (
qcom_sGet *get
)
{
net_sUpdateCircBuffer *mp = get->data;
net_sUpdateCircBufferR *rmp;
gdb_sNode *np;
pwr_tStatus sts;
pwr_tUInt32 size;
pwr_tUInt32 total_size;
qcom_sPut put;
gdh_sCircBuffInfo info[10];
int i, offs;
gdb_ScopeLock {
np = hash_Search(&sts, gdbroot->nid_ht, &mp->hdr.nid);
} gdb_ScopeUnlock;
if (gdbroot->db->log.b.id) {
errh_Info("Sending 'updateCircBuffer' to %s (%s)",
np->name, cdh_NodeIdToString(NULL, np->nid, 0, 0));
}
if ( mp->info_size >= 10) {
errh_Error("Parameter size error 'updateCircBufferR' to %s (%s)",
np->name, cdh_NodeIdToString(NULL, np->nid, 0, 0));
return;
}
for ( i = 0; i < mp->info_size; i++) {
info[i].circ_aref = mp->circ_aref[i];
info[i].resolution = mp->resolution[i];
info[i].samples = mp->samples[i];
info[i].last_idx = mp->last_idx[i];
info[i].offset = mp->offset[i];
info[i].bufsize = mp->bufsize[i];
info[i].bufp = calloc( 1, info[i].bufsize);
}
sts = gdh_UpdateCircBuffInfo( info, mp->info_size);
if ( EVEN(sts))
size = sizeof(*rmp);
else {
total_size = 0;
for ( i = 0; i < mp->info_size; i++)
total_size += info[i].bufsize;
size = sizeof(*rmp) + total_size;
}
size = (size + 3) & ~3; /* Size up to nearest multiple of 4. */
rmp = net_Alloc(&sts, &put, size, net_eMsg_updateCircBufferR);
if (rmp == NULL) {
errh_Error("Failed to allocate 'updateCircBufferR' to %s (%s)",
np->name, cdh_NodeIdToString(NULL, np->nid, 0, 0));
return;
}
rmp->sts = sts;
if ( EVEN(sts)) {
rmp->sts = sts;
rmp->bsize = 0;
}
else {
rmp->sts = sts;
rmp->info_size = mp->info_size;
rmp->bsize = total_size;
offs = 0;
for ( i = 0; i < mp->info_size; i++) {
rmp->circ_aref[i] = info[i].circ_aref;
rmp->size[i] = info[i].size;
rmp->bufsize[i] = info[i].bufsize;
rmp->first_idx[i] = info[i].first_idx;
rmp->last_idx[i] = info[i].last_idx;
rmp->offset[i] = info[i].offset;
memcpy( rmp->buf + offs, info[i].bufp, info[i].bufsize);
offs += info[i].bufsize;
free( info[i].bufp);
}
}
net_Reply(&sts, get, &put, 0);
}
src/lib/rt/src/rt_gdh.c
View file @ a41fe692
......@@ -5076,6 +5076,66 @@ gdh_GetCircBuffInfo (
ap = vol_ArefToAttribute(&sts, &attribute, &info[j].circ_aref, gdb_mLo_global, vol_mTrans_all);
if (ap == NULL || ap->op == NULL) break;
if ( ap->op->l.flags.b.isCached) {
net_sGetCircBuffer *mp;
qcom_sPut put;
gdb_sVolume *vp;
qcom_sQid tgt;
qcom_sGet get;
net_sGetCircBufferR *rsp;
gdb_sNode *np;
vp = pool_Address(NULL, gdbroot->pool, ap->op->l.vr);
if (vp == NULL) {
sts = GDH__NOSUCHOBJ;
break;
}
mp = net_Alloc(&sts, &put, sizeof(*mp), net_eMsg_getCircBuffer);
if (mp == NULL) break;
mp->circ_aref = info->circ_aref;
mp->resolution = info->resolution;
mp->samples = info->samples;
mp->bufsize = info->bufsize;
np = hash_Search(&sts, gdbroot->nid_ht, &vp->g.nid);
if (np == NULL) break;
tgt = np->handler;
gdb_Unlock;
rsp = net_Request(&sts, &tgt, &put, &get, net_eMsg_getCircBufferR, 0, 0);
gdb_Lock;
if (EVEN(sts))
break;
if (EVEN(rsp->sts)) {
sts = rsp->sts;
net_Free(NULL, rsp);
break;
}
if ( cdh_ObjidIsNotEqual(rsp->circ_aref.Objid, info->circ_aref.Objid)) {
sts = 0;
break;
}
info->size = rsp->size;
info->bufsize = rsp->bufsize;
info->first_idx = rsp->first_idx;
info->last_idx = rsp->last_idx;
info->offset = rsp->offset;
info->bufp = calloc( 1, info->bufsize);
memcpy( info->bufp, rsp->buf, info->bufsize);
net_Free(NULL, rsp);
break;
}
touchObject(ap->op);
samples = info[j].samples;
......@@ -5184,6 +5244,7 @@ gdh_UpdateCircBuffInfo (
int split = 0;
int start_idx;
int first_index, last_index;
int finish = 0;
int j;
memset(&attribute, 0, sizeof(attribute));
......@@ -5196,6 +5257,80 @@ gdh_UpdateCircBuffInfo (
ap = vol_ArefToAttribute(&sts, &attribute, &info[j].circ_aref, gdb_mLo_global, vol_mTrans_all);
if (ap == NULL || ap->op == NULL) break;
if ( ap->op->l.flags.b.isCached) {
net_sUpdateCircBuffer *mp;
qcom_sPut put;
gdb_sVolume *vp;
qcom_sQid tgt;
qcom_sGet get;
net_sUpdateCircBufferR *rsp;
gdb_sNode *np;
int i;
int offs;
vp = pool_Address(NULL, gdbroot->pool, ap->op->l.vr);
if (vp == NULL) {
sts = GDH__NOSUCHOBJ;
break;
}
mp = net_Alloc(&sts, &put, sizeof(*mp), net_eMsg_updateCircBuffer);
if (mp == NULL) break;
mp->info_size = infosize;
if ( mp->info_size > 10)
mp->info_size = 10;
for ( i = 0; i < infosize; i++) {
mp->circ_aref[i] = info[i].circ_aref;
mp->resolution[i] = info[i].resolution;
mp->samples[i] = info[i].samples;
mp->bufsize[i] = info[i].bufsize;
mp->last_idx[i] = info[i].last_idx;
mp->offset[i] = info[i].offset;
}
np = hash_Search(&sts, gdbroot->nid_ht, &vp->g.nid);
if (np == NULL) break;
tgt = np->handler;
gdb_Unlock;
rsp = net_Request(&sts, &tgt, &put, &get, net_eMsg_updateCircBufferR, 0, 0);
gdb_Lock;
if (EVEN(sts))
break;
if (EVEN(rsp->sts)) {
sts = rsp->sts;
net_Free(NULL, rsp);
break;
}
offs = 0;
for ( i = 0; i < rsp->info_size; i++) {
if ( cdh_ObjidIsNotEqual(rsp->circ_aref[i].Objid, info[i].circ_aref.Objid)) {
sts = 0;
break;
}
info[i].size = rsp->size[i];
info[i].bufsize = rsp->bufsize[i];
info[i].first_idx = rsp->first_idx[i];
info[i].last_idx = rsp->last_idx[i];
info[i].offset = rsp->offset[i];
info[i].bufp = calloc( 1, info[i].bufsize);
memcpy( info[i].bufp, rsp->buf + offs, info[i].bufsize);
offs += info[i].bufsize;
}
net_Free(NULL, rsp);
finish = 1;
break;
}
touchObject(ap->op);
hp = vol_AttributeToAddress(&sts, ap);
......@@ -5283,6 +5418,9 @@ gdh_UpdateCircBuffInfo (
break;
}
gdh_Unlock;
if ( finish)
break;
}
return sts;
}
src/lib/rt/src/rt_net.x
View file @ a41fe692
......@@ -110,6 +110,11 @@ enum net_eMsg {
net_eMsg_fileList,
net_eMsg_fileListR,
net_eMsg_getCircBuffer,
net_eMsg_getCircBufferR,
net_eMsg_updateCircBuffer,
net_eMsg_updateCircBufferR,
net_eMsg_, /* Not a valid message */
net_eMsg_volumes7, /* Version 7. Internal only */
......@@ -1306,10 +1311,10 @@ struct net_sFileList {
%
%
%typedef struct {
% net_sMessage hdr;/**< Header */
% pwr_tStatus sts;/**< Status */
% pwr_tUInt32 filecnt; /**< Number of files found. */
% char files[1]; /**< List of files */
% net_sMessage hdr; /**< Header */
% pwr_tStatus sts; /**< Status */
% pwr_tUInt32 filecnt; /**< Number of files found. */
% char files[1]; /**< List of files */
%} net_sFileListR;
%
%bool_t xdr_net_sFileListR(XDR *xdrs, net_sFileListR *objp);
......@@ -1357,9 +1362,187 @@ struct net_sFileList {
%}
#endif
struct net_sGetCircBuffer {
net_sMessage hdr;
pwr_sAttrRef circ_aref;
pwr_tUInt32 resolution;
pwr_tUInt32 samples;
pwr_tUInt32 bufsize;
};
#ifdef RPC_HDR
%
%
%typedef struct {
% net_sMessage hdr; /**< Header */
% pwr_tStatus sts; /**< Status */
% pwr_sAttrRef circ_aref; /**< CircBuffer attribute reference. */
% pwr_tUInt32 first_idx; /**< First index in buffer. */
% pwr_tUInt32 last_idx; /**< Last index in buffer. */
% pwr_tUInt32 offset; /**< Offset for next update. */
% pwr_tUInt32 size; /**< Returned number of samples. */
% pwr_tUInt32 bufsize; /**< Size data in bytes. */
% char buf[1]; /**< Data */
%} net_sGetCircBufferR;
%
%bool_t xdr_net_sGetCircBufferR(XDR *xdrs, net_sGetCircBufferR *objp);
%
#elif defined RPC_XDR
%
%bool_t
%xdr_net_sGetCircBufferR(XDR *xdrs, net_sGetCircBufferR *objp)
%{
% pwr_tStatus sts;
% pwr_tUInt32 bufsize;
%
% if (xdrs->x_op == XDR_DECODE) {
% sts = (int) ntohl(objp->sts);
% bufsize = (int) ntohl(objp->bufsize);
% } else {
% sts = objp->sts;
% bufsize = objp->bufsize;
% }
%
% if (!xdr_net_sMessage(xdrs, &objp->hdr)) {
% return (FALSE);
% }
% if (!xdr_pwr_tStatus(xdrs, &objp->sts)) {
% return (FALSE);
% }
%
% if (EVEN(sts)) {
% if (!xdr_pwr_sAttrRef(xdrs, &objp->circ_aref)) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->size )) {
% return (FALSE);
% }
% return (TRUE);
% }
%
%
% if (!xdr_pwr_sAttrRef(xdrs, &objp->circ_aref)) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->size )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->last_idx )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->first_idx )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->offset )) {
% return (FALSE);
% }
% if (!xdr_opaque(xdrs, (char *)objp->buf, bufsize)) {
% return (FALSE);
% }
% return (TRUE);
%
%}
#endif
struct net_sUpdateCircBuffer {
net_sMessage hdr;
pwr_tUInt32 info_size;
pwr_sAttrRef circ_aref[10];
pwr_tUInt32 resolution[10];
pwr_tUInt32 samples[10];
pwr_tUInt32 last_idx[10];
pwr_tUInt32 offset[10];
pwr_tUInt32 bufsize[10];
};
#ifdef RPC_HDR
%
%
%typedef struct {
% net_sMessage hdr; /**< Header */
% pwr_tStatus sts; /**< Status */
% pwr_tUInt32 info_size;
% pwr_sAttrRef circ_aref[10]; /**< CircBuffer attribute reference. */
% pwr_tUInt32 first_idx[10]; /**< First index in buffer. */
% pwr_tUInt32 last_idx[10]; /**< Last index in buffer. */
% pwr_tUInt32 offset[10]; /**< Offset for next update. */
% pwr_tUInt32 size[10]; /**< Returned number of samples. */
% pwr_tUInt32 bufsize[10]; /**< Size data for each item in bytes. */
% pwr_tUInt32 bsize; /**< Total size data in bytes. */
% char buf[1]; /**< Data */
%} net_sUpdateCircBufferR;
%
%bool_t xdr_net_sUpdateCircBufferR(XDR *xdrs, net_sUpdateCircBufferR *objp);
%
#elif defined RPC_XDR
%
%bool_t
%xdr_net_sUpdateCircBufferR(XDR *xdrs, net_sUpdateCircBufferR *objp)
%{
% pwr_tStatus sts;
% pwr_tUInt32 bsize;
% pwr_tUInt32 info_size;
% int i;
%
% if (xdrs->x_op == XDR_DECODE) {
% sts = (int) ntohl(objp->sts);
% bsize = (int) ntohl(objp->bsize);
% info_size = (int) ntohl(objp->info_size);
% } else {
% sts = objp->sts;
% bsize = objp->bsize;
% info_size = objp->info_size;
% }
%
% if (!xdr_net_sMessage(xdrs, &objp->hdr)) {
% return (FALSE);
% }
% if (!xdr_pwr_tStatus(xdrs, &objp->sts)) {
% return (FALSE);
% }
%
% if (EVEN(sts)) {
% if (!xdr_pwr_tUInt32(xdrs, &objp->info_size )) {
% return (FALSE);
% }
% return (TRUE);
% }
%
%
% if (!xdr_pwr_tUInt32(xdrs, &objp->bsize )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->info_size )) {
% return (FALSE);
% }
% for ( i = 0; i < info_size; i++) {
% if (!xdr_pwr_sAttrRef(xdrs, &objp->circ_aref[i])) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->last_idx[i] )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->first_idx[i] )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->offset[i] )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->size[i] )) {
% return (FALSE);
% }
% if (!xdr_pwr_tUInt32(xdrs, &objp->bufsize[i] )) {
% return (FALSE);
% }
% }
% if (!xdr_opaque(xdrs, (char *)objp->buf, bsize)) {
% return (FALSE);
% }
% return (TRUE);
%
%}
#endif
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7