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

GATFileStream.c Go to the documentation of this file. /** @file GATFileStream.c * Main .c for the GATFileStream class. * * An GATFileStream represents a connection to another process. * * @date $Date: 2004/04/09 12:29:18 $ * * @version $Header: /export/cvs-gridlab/wp-1/Codes/GATEngine/C-reference/src/GATFileStream.c,v 1.11 2004/04/09 12:29:18 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/GATFileStream.c,v 1.11 2004/04/09 12:29:18 hartmutkaiser Exp $"; /* System Header Files */ #include <stdio.h> #include <stdlib.h> #include <string.h> #include <assert.h> /* GAT Header Files */ #include "GAT.h" #include "GATRegistry.h" #include "GATInternal.h" #include "GATFileStreamCPI.h" #include "GATMetricEvent.h" /* define the vtable types */ GATOBJECT_DEFINE_VTABLE(GATFileStream); GATSTREAMABLE_DEFINE_VTABLE(GATFileStream); GATMONITORABLE_DEFINE_VTABLE(GATFileStream); /* Declare the converters to/from GATObject */ GATOBJECT_DEFINE_CONVERTERS(GATFileStream); /* Macros */ /* Structures, unions and enums */ struct GATFileStream_S { /* supported interfaces */ GATFileStream_vtable *GATObject__vtable; GATFileStream_IStreamable_vtable *GATStreamable__vtable; GATFileStream_IMonitorable_vtable *GATMonitorable__vtable; /* instance data */ GATFileStreamCPI_Instance data; /* instance data of the corresponding CPI object */ GATFileStreamCPI cpi; /* CPI object to use for all subsequent operations */ GATFileStreamCPIList cpilist; /* CPI list of appropriate CPI objects */ }; /* Static function prototypes */ static GATResult GATFileStream_GetCPIInstanceData(GATFileStream filestream, void **data); /* FileStream scope variables */ static GATFileStream_vtable GATFileStream__vtable = { GATFileStream_GetType, GATFileStream_Destroy, GATFileStream_Equals, GATFileStream_Clone, GATFileStream_GetInterface, GATFileStream_GetCPIInstanceData }; static GATFileStream_IStreamable_vtable GATFileStream_IStreamable__vtable = { GATFileStream_Read, GATFileStream_Write, GATFileStream_Seek, GATFileStream_Close }; static GATFileStream_IMonitorable_vtable GATFileStream_IMonitorable__vtable = { GATFileStream_AddMetricListener, GATFileStream_RegisterPolling, GATFileStream_RemoveRegisteredMetric, GATFileStream_GetMetrics }; /* External functions */ /** * Constructs an instance of this class. * * @param context Used to broker resources. * @param preferences User preferences for this instance. * @param location The location of the file to open * @param mode The mode in which to open the file (GATLogicalFileMode_Read, * GATLogicalFileMode_Write, GATLogicalFileMode_Append, or GATLogicalFileMode_ReadWrite). */ GATFileStream GATFileStream_Create(GATContext context, GATPreferences_const preferences, GATLocation location, GATFileStreamMode mode) { GAT_STATUS_APIENTRY(context, "GATFileStream_Create"); GATFileStream new_fileStream = (GATFileStream) malloc(sizeof(struct GATFileStream_S)); if(NULL != new_fileStream) { memset(new_fileStream, 0, sizeof(struct GATFileStream_S)); new_fileStream->GATObject__vtable = &GATFileStream__vtable; new_fileStream->GATStreamable__vtable = &GATFileStream_IStreamable__vtable; new_fileStream->GATMonitorable__vtable = &GATFileStream_IMonitorable__vtable; new_fileStream->data.mode = mode; new_fileStream->data.context = context; new_fileStream->data.source = GATFileStream_ToGATObject_const(new_fileStream); /* find a CPI object, which can handle the fileStream */ { GATRegistry_const registry = GATContext_internal_GetRegistry(context); GAT_CREATE_STATUS(GATLocation_Clone(location, &new_fileStream->data.location)); /* use default preferences, if appropriate */ if (NULL == preferences) { preferences = GATContext_GetPreferences(context); } new_fileStream->cpilist = GATRegistry_FindGATFileStreamCPI(registry, preferences); if (NULL == new_fileStream->cpilist) { GAT_CREATE_STATUS(GAT_NO_REGISTERED_CPI); } else if (GAT_SUCCEEDED(GAT_CURRENT_STATUS())) { /* create a new instance of the CPI provider */ GATBool found_cpi = GATFalse; GATFileStreamCPIList current = NULL; for(current = new_fileStream->cpilist; NULL != current; current = current->next) { GAT_CREATE_STATUS(GATFileStreamCPI_CreateInstance(current->cpi, &new_fileStream->data)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS())) { new_fileStream->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 = GATFileStreamCPI_GetMetrics(new_fileStream->cpi, &new_fileStream->data, &metrics); if (GAT_SUCCEEDED(retval)) { new_fileStream->data.monitorable = GATMonitorable_Impl_Create(metrics); if (NULL == new_fileStream->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_AddGATFileStreamToCPIList(context, new_fileStream->cpi, new_fileStream)); } } } } else { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } /* error handling */ if (GAT_FAILED(GAT_CURRENT_STATUS())) { GATFileStream_Destroy(&new_fileStream); } GAT_STORE_STATUS(); return new_fileStream; } /** GATFileStream_Destroy * The GATFileStream destructor. * This is the destructor for GATFileStream objects. * * @param fileStream An old GATFileStream */ void GATFileStream_Destroy(GATFileStream *fileStream) { if(NULL != fileStream && NULL != *fileStream) { GATMonitorable_Impl_Destroy(&(*fileStream)->data.monitorable); if (NULL != (*fileStream)->cpi) { GATRegistry_RemoveGATFileStreamFromCPIList((*fileStream)->data.context, (*fileStream)->cpi, *fileStream); GATFileStreamCPI_DestroyInstance((*fileStream)->cpi, &(*fileStream)->data); } GATLocation_Destroy(&(*fileStream)->data.location); GATFileStreamCPIList_Destroy((*fileStream)->cpilist); free(*fileStream); *fileStream = NULL; } } /** int GATFileStream_GetType(GATFileStream_const fileStream) * @brief Return the type of the GATFileStream * * The function @c GATFileStream_GetType always returns GATType_GATFileStream. * * @param fileStream The object to inspect * * @return returns always @c #GATType_GATFileStream. */ GATType GATFileStream_GetType(GATFileStream_const fileStream) { GAT_UNUSED_PARAMETER(fileStream); return GATType_GATFileStream; } /** int GATFileStream_Clone(GATFileStream_const fileStream, GATFileStream *new_fileStream) * @brief Clone the given GATFileStream * * The function @c GATFileStream_Clone generates a (deep) copy of the given * GATFileStream. * * @param new_object The object to clone * @param new_fileStream The pointer, through which the result is to be * returned. * * @return An error type. */ GATResult GATFileStream_Clone(GATFileStream_const fileStream, GATFileStream *new_object) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_Clone"); if (NULL != new_object) { struct GATFileStream_S const *passedFileStream = fileStream; GATFileStream new_fileStream = (GATFileStream) malloc(sizeof(struct GATFileStream_S)); *new_object = NULL; if (NULL == new_fileStream) { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } else { memset(new_fileStream, 0, sizeof(struct GATFileStream_S)); new_fileStream->GATObject__vtable = &GATFileStream__vtable; new_fileStream->GATStreamable__vtable = &GATFileStream_IStreamable__vtable; new_fileStream->GATMonitorable__vtable = &GATFileStream_IMonitorable__vtable; new_fileStream->data.mode = passedFileStream->data.mode; new_fileStream->data.context = passedFileStream->data.context; new_fileStream->data.source = GATFileStream_ToGATObject_const(new_fileStream); new_fileStream->cpilist = GATRegistry_CloneGATFileStreamCPIList(passedFileStream->cpilist); if (NULL == new_fileStream->cpilist) { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } GAT_CREATE_STATUS(GATLocation_Clone(passedFileStream->data.location, &new_fileStream->data.location)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS())) { /* re-find the cpi to use */ GATFileStreamCPIList current = passedFileStream->cpilist; GATFileStreamCPIList new_current = new_fileStream->cpilist; for (/**/; NULL != current; current = current->next, new_current = new_current->next) { if (current->cpi == passedFileStream->cpi) { new_fileStream->cpi = new_current->cpi; break; } } assert(NULL != new_fileStream->cpi); /* clone the instance data */ GAT_CREATE_STATUS(GATFileStreamCPI_CloneInstance(passedFileStream->cpi, &passedFileStream->data, &new_fileStream->data)); /* clone the monitoring data */ if (NULL != passedFileStream->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_Clone(passedFileStream->data.monitorable, &new_fileStream->data.monitorable)); } /* register this object with the engine */ GAT_CREATE_STATUS(GATRegistry_AddGATFileStreamToCPIList( new_fileStream->data.context, new_fileStream->cpi, new_fileStream)); } } /* destroy the newly allocated object, if some error occured */ if (GAT_FAILED(GAT_CURRENT_STATUS())) { GATFileStream_Destroy(&new_fileStream); } else { /* return the new object */ *new_object = new_fileStream; } } else { GAT_CREATE_STATUS(GAT_MEMORYFAILURE); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** int GATFileStream_Equals(GATFileStream_const lhs, GATFileStream_const rhs, GATBool *isequal) * @brief Compares two GATFileStream objects * * The function @c GATFileStream_Equals compares two fileStream objects of type * @c #GATFileStream. * * @param lhs The first fileStream object to compare * @param rhs The second fileStream object to compare * @param isequal The pointer to the GATBool variable, which should receive * the result if the comparision * * @return An error code. */ GATResult GATFileStream_Equals(GATFileStream_const lhs, GATFileStream_const rhs, GATBool *isequal) { if (NULL != lhs && NULL != rhs) { GAT_STATUS_APIENTRY(lhs->data.context, "GATFileStream_Equals"); if (NULL != isequal) { GAT_CREATE_STATUS(GATLocation_Equals(lhs->data.location, rhs->data.location, isequal)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS()) && GATTrue == *isequal) { if (lhs->data.mode != rhs->data.mode) { *isequal = GATFalse; } if (GATTrue == *isequal) { GAT_CREATE_STATUS(GATFileStreamCPI_EqualsInstance(lhs->cpi, &lhs->data, &rhs->data, isequal)); } } } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /** int GATFileStream_GetInterface(GATFileStream_const fileStream, GATInterface iftype, void const **ifp) * @brief Get an interface supported by a GATObject * * The function GATFileStream_GetInterface allows to get a pointer to an * additional interface supported by this GATFileStream. * * @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 GATFileStream_GetInterface(GATFileStream_const fileStream, GATInterface iftype, void const **ifp) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_GetInterface"); if (NULL != ifp) { if (GATInterface_IStreamable == iftype) { *ifp = (void const *) &fileStream->GATStreamable__vtable; } else if (GATInterface_IMonitorable == iftype && NULL != fileStream->data.monitorable) { *ifp = (void const *) &fileStream->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 */ /** GATFileStream_AddMetricListener * * The function GATFileStream_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 fileStream The GATFileStream 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 GATFileStream_AddMetricListener(GATFileStream fileStream, GATMetricListener listener, void *listener_data, GATMetric metric, GATuint32 *cookie) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_AddMetricListener"); if (NULL != fileStream->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_AddMetricListener(fileStream->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; } /** GATFileStream_RegisterPolling * * The function GATFileStream_RegisterPolling registers a continuous * metric with the given GATFileStream instance. * * @param fileStream The GATFileStream 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 GATFileStream. * * @return An error code. */ GATResult GATFileStream_RegisterPolling(GATFileStream fileStream, GATMetric metric, GATMetricEvent *event, GATuint32 *cookie) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_RegisterPolling"); if (NULL != fileStream->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_RegisterPolling(fileStream->data.monitorable, metric, 0, cookie)); /* ask the CPI provider to return the corresponding metric event */ GAT_CREATE_STATUS(GATFileStreamCPI_GetMetricEvent(fileStream->cpi, &fileStream->data, metric, event)); if (GAT_SUCCEEDED(GAT_CURRENT_STATUS()) && NULL != cookie && 0 != *cookie) { GATMonitorable_Impl_RemoveRegisteredMetric(fileStream->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; } /** GATFileStream_RemoveRegisteredMetric * * The function GATFileStream_RemoveRegisteredMetric removes a GATMetricListener, * which was previously registered with GATFileStream_AddMetricListener. * * @param fileStream The GATFileStream 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 GATFileStream_RemoveRegisteredMetric(GATFileStream fileStream, GATMetric metric, GATuint32 cookie) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_RemoveRegisteredMetric"); if (NULL != fileStream->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_RemoveRegisteredMetric( fileStream->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; } /** GATFileStream_GetMetrics * * The function GATFileStream_GetMetrics returns a list of metrics supported by this * GATFileStream. * * @param fileStream The GATFileStream 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 GATFileStream_GetMetrics(GATFileStream_const fileStream, GATList_GATMetric *metrics) { if (NULL != fileStream) { GAT_STATUS_APIENTRY(fileStream->data.context, "GATFileStream_GetMetrics"); if (NULL != fileStream->data.monitorable) { GAT_CREATE_STATUS(GATMonitorable_Impl_GetMetrics( fileStream->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 GATFileStream_Read(GATFileStream object, void *buffer, GATuint32 size, GATuint32 *read_bytes) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATFileStream_Read"); GAT_CREATE_STATUS(GATFileStreamCPI_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 GATFileStream_Write(GATFileStream object, void const *buffer, GATuint32 size, GATuint32 *written_bytes) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATFileStream_Write"); GAT_CREATE_STATUS(GATFileStreamCPI_Write(object->cpi, &object->data, buffer, size, written_bytes)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* reposition the internal stream position */ GATResult GATFileStream_Seek(GATFileStream object, GATOrigin origin, GATint32 offset, GATuint32 *new_position) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATFileStream_Seek"); GAT_CREATE_STATUS(GATFileStreamCPI_Seek(object->cpi, &object->data, origin, offset, new_position)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* Closes this GATStreamable instance. */ GATResult GATFileStream_Close(GATFileStream object) { if (NULL != object) { GAT_STATUS_APIENTRY(object->data.context, "GATFileStream_Close"); GAT_CREATE_STATUS(GATFileStreamCPI_Close(object->cpi, &object->data)); return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; } /* Local functions */ /* GATFileStream_GetCPIInstanceData * * The function GATFileStream_GetCPIInstanceData returns the CPI instance data, * associated with the given GATFile object. * * @param filestream 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 GATFileStream_GetCPIInstanceData(GATFileStream filestream, void **data) { if (NULL != filestream) { GAT_USES_STATUS(filestream->data.context, "GATFileStream_GetCPIInstanceData"); if (NULL != data) { *data = (void *)&filestream->data; } else { GAT_CREATE_STATUS(GAT_INVALID_PARAMETER); } return GAT_RETURN_STATUS(); } return GAT_INVALID_HANDLE; }