Grid Application Toolkit A simple API for Grid Applications

Menu Home API-Specification API-Overview Status C Reference Manual Download Java Java-Reference General GridLab - WP-1 Links

GATPipe.c Go to the documentation of this file. /** @file GATPipe.c * Main .c for the GATPipe class. * * An GATPipe represents a connection to another process. * * @date $Date: 2004/04/20 12:33:22 $ * * @version $Header: /export/cvs-gridlab/wp-1/Codes/GATEngine/C-reference/src/GATPipe.c,v 1.12 2004/04/20 12:33:22 hartmutkaiser Exp $ * * Copyright (C) Kelly Davis * This is part of the GAT Engine. * Contributed by Kelly Davis <kdavis@aei.mpg.de> * * Use, modification and distribution is subject to the Gridlab Software * License. (See accompanying file GLlicense.txt or copy at * http://www.gridlab.org/GLlicense.txt) */ static const char *rcsid = "$Header: /export/cvs-gridlab/wp-1/Codes/GATEngine/C-reference/src/GATPipe.c,v 1.12 2004/04/20 12:33:22 hartmutkaiser Exp $"; /* System Header Files */ #include <stdio.h> #include <stdlib.h> #include <string.h> #include <assert.h> /* GAT Header Files */ #include "GAT.h" #include "GATPipe.h" #include "GATPipeCPI.h" #include "GATRegistry.h" #include "GATInternal.h" #include "GATXdsWrapper.h" #include "GATMetricEvent.h" /* define the vtable types */ GATOBJECT_DEFINE_VTABLE(GATPipe); GATSTREAMABLE_DEFINE_VTABLE(GATPipe); GATMONITORABLE_DEFINE_VTABLE(GATPipe); /* Declare the converters to/from GATObject */ GATOBJECT_DEFINE_CONVERTERS(GATPipe); /* Macros */ /* Structures, unions and enums */ struct GATPipe_S { /* supported interfaces */ GATPipe_vtable *GATObject__vtable; GATPipe_IStreamable_vtable *GATStreamable__vtable; GATPipe_IMonitorable_vtable *GATMonitorable__vtable; /* instance data */ GATPipeCPI_Instance data; /* instance data of the corresponding CPI object */ GATPipeCPI cpi; /* CPI object to use for all subsequent operations */ GATPipeCPIList cpilist; /* CPI list of appropriate CPI objects */ }; /* Static function prototypes */ static GATResult GATPipe_GetCPIInstanceData(GATPipe pipe, void **data); /* Pipe scope variables */ static GATPipe_vtable GATPipe__vtable = { GATPipe_GetType, GATPipe_Destroy, GATPipe_Equals, GATPipe_Clone, GATPipe_GetInterface, GATPipe_GetCPIInstanceData }; static GATPipe_IStreamable_vtable GATPipe_IStreamable__vtable = { GATPipe_Read, GATPipe_Write, GATPipe_Seek, GATPipe_Close }; static GATPipe_IMonitorable_vtable GATPipe_IMonitorable__vtable = { GATPipe_AddMetricListener, GATPipe_RegisterPolling, GATPipe_RemoveRegisteredMetric, GATPipe_GetMetrics }; /* External functions */ /** GATPipe_Create * Constructs an instance of this class. * * @param context Used to broker resources. * @param preferences User preferences for this instance. * @param information Other information required to construct a GATPipe. */ GATPipe GATPipe_Create(GATContext context, GATPreferences_const preferences, void *information) { GAT_STATUS_APIENTRY(context, "GATPipe_Create"); GATPipe new_pipe = (GATPipe) malloc(sizeof(struct GATPipe_S)); if (NULL != new_pipe) { memset(new_pipe, 0, sizeof(struct GATPipe_S)); new_pipe->GATObject__vtable = &GATPipe__vtable; new_pipe->GATStreamable__vtable = &GATPipe_IStreamable__vtable; new_pipe->GATMonitorable__vtable = &GATPipe_IMonitorable__vtable; new_pipe->data.context = context; new_pipe->data.source = GATPipe_ToGATObject_const(new_pipe); /* find a CPI object, which can handle the pipe */ { GATRegistry_const registry = GATContext_internal_GetRegistry(context); /* use default preferences, if appropriate */ if (NULL == preferences) { preferences = GATContext_GetPreferences(context); } new_pipe->cpilist = GATRegistry_FindGATPipeCPI(registry, preferences); if (NULL == new_pipe->cpilist) { GAT_CREATE_STATUS(GAT_NO_REGISTERED_CPI); } else { /* create a new instance of the CPI provider */ GATBool found_cpi = GATFalse; GATPipeCPIList current = NULL; for(current = new_pipe->cpilist; NULL != current; current = current->next) { GAT_CREATE_STATUS(GATPipeCPI_CreateInstance(current->cpi, &new_pipe->data, information)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS())) { new_pipe->cpi = current->cpi; found_cpi = GATTrue; break; } } /* no available CPI object can handle this */ if (GATFalse == found_cpi) { GAT_CREATE_STATUS(GAT_NO_REGISTERED_CPI); } else { /* now, initialise the monitoring framework with the help of the CPI provider found */ GATList_GATMetric metrics = NULL; GATResult retval = GATPipeCPI_GetMetrics(new_pipe->cpi, &new_pipe->data, &metrics); if (GAT_SUCCEEDED(retval)) { new_pipe->data.monitorable = GATMonitorable_Impl_Create(metrics); if (NULL == new_pipe->data.monitorable) { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } GATList_GATMetric_Destroy(&metrics); } else if (GAT_NOTIMPL == retval) { retval = GAT_SUCCESS; /* no monitoring supported here */ } else { GAT_CREATE_STATUS(retval); } /* register this object with the engine */ GAT_CREATE_STATUS(GATRegistry_AddGATPipeToCPIList(context, new_pipe->cpi, new_pipe)); } } } } else { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } /* error handling */ if (GAT_FAILED(GAT_CURRENT_STATUS())) { GATPipe_Destroy(&new_pipe); } GAT_STORE_STATUS(); return new_pipe; } /** GATPipe_Destroy * The GATPipe destructor. * This is the destructor for GATPipe objects. * * @param peep An old GATPipe */ void GATPipe_Destroy(GATPipe *peep) { if(NULL != peep && NULL != *peep) { GATMonitorable_Impl_Destroy(&(*peep)->data.monitorable); if (NULL != (*peep)->cpi) { GATRegistry_RemoveGATPipeFromCPIList((*peep)->data.context, (*peep)->cpi, *peep); GATPipeCPI_DestroyInstance((*peep)->cpi, &(*peep)->data); } GATPipeCPIList_Destroy((*peep)->cpilist); free(*peep); *peep = NULL; } } /** int GATPipe_GetType(GATPipe_const peep) * @brief Return the type of the GATPipe * * The function @c GATPipe_GetType always returns GATType_GATPipe. * * @param peep The object to inspect * * @return returns always @c #GATType_GATPipe. */ GATType GATPipe_GetType(GATPipe_const peep) { GAT_UNUSED_PARAMETER(peep); return GATType_GATPipe; } /** int GATPipe_Clone(GATPipe_const peep, GATPipe *new_peep) * @brief Clone the given GATPipe * * The function @c GATPipe_Clone generates a (deep) copy of the given * GATPipe. * * @param new_object The object to clone * @param new_peep The pointer, through which the result is to be * returned. * * @return An error type. */ GATResult GATPipe_Clone(GATPipe_const pipe, GATPipe *new_object) { if (NULL != pipe) { GAT_STATUS_APIENTRY(pipe->data.context, "GATPipe_Clone"); if (NULL != new_object) { struct GATPipe_S const *passedPipe = pipe; GATPipe new_pipe = (GATPipe) malloc(sizeof(struct GATPipe_S)); *new_object = NULL; if (NULL == new_pipe) { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } else { memset(new_pipe, 0, sizeof(struct GATPipe_S)); new_pipe->GATObject__vtable = &GATPipe__vtable; new_pipe->GATStreamable__vtable = &GATPipe_IStreamable__vtable; new_pipe->GATMonitorable__vtable = &GATPipe_IMonitorable__vtable; new_pipe->data.context = passedPipe->data.context; new_pipe->data.source = GATPipe_ToGATObject_const(new_pipe); new_pipe->cpilist = GATRegistry_CloneGATPipeCPIList(passedPipe->cpilist); if (NULL == new_pipe->cpilist) { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } else { /* re-find the cpi to use */ GATPipeCPIList current = passedPipe->cpilist; GATPipeCPIList new_current = new_pipe->cpilist; for (/**/; NULL != current; current = current->next, new_current = new_current->next) { if (current->cpi == passedPipe->cpi) { new_pipe->cpi = new_current->cpi; break; } } assert(NULL != new_pipe->cpi); /* clone the instance data */ GAT_CREATE_STATUS(GATPipeCPI_CloneInstance(passedPipe->cpi, &passedPipe->data, &new_pipe->data)); /* clone the monitoring data */ if (NULL != passedPipe->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_Clone(passedPipe->data.monitorable, &new_pipe->data.monitorable)); } GAT_CREATE_STATUS(GATRegistry_AddGATPipeToCPIList(new_pipe->data.context, new_pipe->cpi, new_pipe)); } } /* destroy the newly allocated object, if some error occured */ if (GAT_FAILED(GAT_CURRENT_STATUS())) { GATPipe_Destroy(&new_pipe); } else { /* return the new object */ *new_object = new_pipe; } } else { GAT_CREATE_STATUS(GAT_INVALID_PARAMETER); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** int GATPipe_Equals(GATPipe_const lhs, GATPipe_const rhs, GATBool *isequal) * @brief Compares two GATPipe objects * * The function @c GATPipe_Equals compares two peep objects of type * @c #GATPipe. * * @param lhs The first peep object to compare * @param rhs The second peep object to compare * @param isequal The pointer to the GATBool variable, which should receive * the result if the comparision * * @return An error code. */ GATResult GATPipe_Equals(GATPipe_const lhs, GATPipe_const rhs, GATBool *isequal) { if (NULL != lhs && NULL != rhs) { GAT_STATUS_APIENTRY(lhs->data.context, "GATPipe_Equals"); if (NULL != isequal) { GAT_CREATE_STATUS(GATPipeCPI_EqualsInstance(lhs->cpi, &lhs->data, &rhs->data, isequal)); } else { GAT_CREATE_STATUS(GAT_INVALID_PARAMETER); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** int GATPipe_GetInterface(GATPipe_const peep, GATInterface iftype, void const **ifp) * @brief Get an interface supported by a GATObject * * The function GATPipe_GetInterface allows to get a pointer to an * additional interface supported by this GATPipe. * * @param object The object to be asked for the new interface. * @param iftype The interface the object is to be asked for. * @param ifp The pointer, through which the result is to be returned. * * @return An error type. */ GATResult GATPipe_GetInterface(GATPipe_const peep, GATInterface iftype, void const **ifp) { if (NULL != peep) { GAT_STATUS_APIENTRY(peep->data.context, "GATPipe_GetInterface"); if (NULL != ifp) { if (GATInterface_IStreamable == iftype) { *ifp = (void const *) &peep->GATStreamable__vtable; } else if (GATInterface_IMonitorable == iftype && NULL != peep->data.monitorable) { *ifp = (void const *) &peep->GATMonitorable__vtable; } else { *ifp = NULL; GAT_CREATE_STATUS(GAT_NO_INTERFACE); } } else { GAT_CREATE_STATUS(GAT_INVALID_PARAMETER); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* GATMonitorable */ /** GATPipe_AddMetricListener * * The function GATPipe_AddMetricListener adds the passed instance of a * GATMetricListener to the list of GATMetricListeners which are notified of * fired GATMetricEvents of the type described by the provided GATMetric * instance. * * @param peep The GATPipe instance to add the GATMetricListener to. * @param listener The GTAMetricListener to call, when the corresponding * GATMetricEvent is fired. * @param listener_data The client supplied data, which will be passed through * to the GATMetricListener. * @param metric The GATMetric instance describing the GATMetricEvent to pass * to the GATMetricListener. * @param cookie The returned value should be used to remove the * GATMetricListener from this GATMonitorable. * * @return An error code. */ GATResult GATPipe_AddMetricListener(GATPipe peep, GATMetricListener listener, void *listener_data, GATMetric metric, GATuint32 *cookie) { if (NULL != peep) { GAT_STATUS_APIENTRY(peep->data.context, "GATPipe_AddMetricListener"); if (NULL != peep->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_AddMetricListener( peep->data.monitorable, listener, listener_data, metric, cookie)); } else { /* this may happen, if the adaptor has not registered any GATMetrics */ GAT_CREATE_STATUS(GAT_NO_INTERFACE); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** GATPipe_RegisterPolling * * The function GATPipe_RegisterPolling registers a continuous * metric with the given GATPipe instance. * * @param peep The GATPipe instance to register the metric with. * @param metric The GATMetric instance describing the GATMetricEvent to pass * to the GATMetricListener. This metric instance must be of type * GATMeasurementType_Continuous, otherwise an error is returned. * @param event The pointer to a variable, which should receive the resulting * GATMetricEvent. The GATMetricEvent instance should be freed by the * caller when it isn't used anymore. * @param cookie The returned value should be used to remove the * GATMetricListener from this GATPipe. * * @return An error code. */ GATResult GATPipe_RegisterPolling(GATPipe peep, GATMetric metric, GATMetricEvent *event, GATuint32 *cookie) { if (NULL != peep) { GAT_STATUS_APIENTRY(peep->data.context, "GATPipe_RegisterPolling"); if (NULL != peep->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_RegisterPolling(peep->data.monitorable, metric, 0, cookie)); /* ask the CPI provider to return the corresponding metric event */ GAT_CREATE_STATUS(GATPipeCPI_GetMetricEvent(peep->cpi, &peep->data, metric, event)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS()) && NULL != cookie && 0 != *cookie) { GATMonitorable_Impl_RemoveRegisteredMetric(peep->data.monitorable, metric, *cookie); *cookie = 0; } } else { /* this may happen, if the adaptor has not registered any GATMetrics */ GAT_CREATE_STATUS(GAT_NO_INTERFACE); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** GATPipe_RemoveRegisteredMetric * * The function GATPipe_RemoveRegisteredMetric removes a GATMetricListener, * which was previously registered with GATPipe_AddMetricListener. * * @param peep The GATPipe instance to remove the GATMetricListener from. * @param metric The GATMetric instance describing the GATMetricEvent to pass * to the GATMetricListener. * @param cookie This value identifies the GATMetricListener to remove, it * was returned from the corresponding GATMonitorable_AddMetricListener * or GATMonitorable_MetricRegisterPolling functions. * * @return An error code. */ GATResult GATPipe_RemoveRegisteredMetric(GATPipe peep, GATMetric metric, GATuint32 cookie) { if (NULL != peep) { GAT_STATUS_APIENTRY(peep->data.context, "GATPipe_RemoveRegisteredMetric"); if (NULL != peep->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_RemoveRegisteredMetric( peep->data.monitorable, metric, cookie)); } else { /* this may happen, if the adaptor has not registered any GATMetrics */ GAT_CREATE_STATUS(GAT_NO_INTERFACE); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** GATPipe_GetMetrics * * The function GATPipe_GetMetrics returns a list of metrics supported by this * GATPipe. * * @param peep The GATPipe instance, for which the supported metrics should be * returned. * @param metrics The pointer to the variable, which receives the resulting * list of metrics. * * @return An error code. */ GATResult GATPipe_GetMetrics(GATPipe_const peep, GATList_GATMetric *metrics) { if (NULL != peep) { GAT_STATUS_APIENTRY(peep->data.context, "GATPipe_GetMetrics"); if (NULL != peep->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_GetMetrics(peep->data.monitorable, metrics)); } else { /* this may happen, if the adaptor has not registered any GATMetrics */ GAT_CREATE_STATUS(GAT_NO_INTERFACE); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* IStreamable implementation */ /* Reads from this GATStreamable into the given buffer */ GATResult GATPipe_Read(GATPipe object, void *buffer, GATuint32 size, GATuint32 *read_bytes) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATPipe_Read"); GAT_CREATE_STATUS(GATPipeCPI_Read(object->cpi, &object->data, buffer, size, read_bytes)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* Writes data from the given Buffer through the GATStreamable */ GATResult GATPipe_Write(GATPipe object, void const *buffer, GATuint32 size, GATuint32 *written_bytes) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATPipe_Write"); GAT_CREATE_STATUS(GATPipeCPI_Write(object->cpi, &object->data, buffer, size, written_bytes)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* reposition the internal stream position */ GATResult GATPipe_Seek(GATPipe object, GATOrigin origin, GATint32 offset, GATuint32 *new_position) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATPipe_Seek"); GAT_CREATE_STATUS(GATPipeCPI_Seek(object->cpi, &object->data, origin, offset, new_position)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* Closes this GATStreamable instance. */ GATResult GATPipe_Close(GATPipe object) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATPipe_Close"); GAT_CREATE_STATUS(GATPipeCPI_Close(object->cpi, &object->data)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* Local functions */ /* GATPipe_GetCPIInstanceData * * The function GATPipe_GetCPIInstanceData returns the CPI instance data, * associated with the given GATPipe object. * * @param object The object, for which to return the CPI instance data. * @param data The pointer to the variable, which should receive the * resulting CPI instance data. * * @return An error code. */ static GATResult GATPipe_GetCPIInstanceData(GATPipe object, void **data) { if (NULL != object) { GAT_USES_STATUS(object->data.context, "GATPipe_GetCPIInstanceData"); if (NULL != data) { *data = (void *)&object->data; } else { GAT_CREATE_STATUS(GAT_INVALID_PARAMETER); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; }