MCPcopy Create free account
hub / github.com/RedisGears/RedisGears / RedisGearsPy_InnerExecute

Function RedisGearsPy_InnerExecute

plugins/python/redisgears_python.c:4782–4867  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

4780}
4781
4782static int RedisGearsPy_InnerExecute(RedisModuleCtx* rctx, BackgroundDepsInstallCtx* bdiCtx){
4783 PythonThreadCtx* ptctx = GetPythonThreadCtx();
4784 ptctx->currentCtx = rctx;
4785 ptctx->createdExecution = NULL;
4786
4787 PythonExecutionCtx pectx = PythonExecutionCtx_New(bdiCtx->session, NULL);
4788 RedisGearsPy_Lock(&pectx);
4789
4790 if (bdiCtx->session->isInstallationNeeded) {
4791 /* Requirements was installed, reset modules catch. */
4792 PyInterpreterState *interp = PyThreadState_GET()->interp;
4793 PyDict_Clear(interp->modules);
4794 PyDict_Merge(interp->modules, initialModulesDict, 1);
4795 }
4796
4797 PyObject *v = PyRun_StringFlags(bdiCtx->script, Py_file_input, ptctx->currSession->globalsDict, ptctx->currSession->globalsDict, NULL);
4798
4799 if(!v){
4800 char* err = getPyError();
4801 if(!err){
4802 RedisModule_ReplyWithError(rctx, "failed running the given script");
4803 }else{
4804 RedisModule_ReplyWithError(rctx, err);
4805 RG_FREE(err);
4806 }
4807
4808 if(ptctx->createdExecution){
4809 // error occured, we need to abort the created execution.
4810 int res = RedisGears_AbortExecution(ptctx->createdExecution);
4811 RedisModule_Assert(res == REDISMODULE_OK);
4812 RedisGears_DropExecution(ptctx->createdExecution);
4813 }
4814
4815 RedisGearsPy_Unlock(&pectx);
4816
4817 ptctx->createdExecution = NULL;
4818 ptctx->currentCtx = NULL;
4819 return REDISMODULE_ERR;
4820 }
4821
4822 if (array_len(bdiCtx->session->registrations) > 0){
4823 // we have registrations, lets register them!
4824 RedisModuleBlockedClient* bc = bdiCtx->bc;
4825 bdiCtx->bc = NULL; // we are taking ownership of the blocked client
4826 if(!bc){
4827 bc = RedisModule_BlockClient(ptctx->currentCtx, NULL, NULL, NULL, 0);
4828 }
4829 char *err = NULL;
4830 if (RedisGears_Register(bdiCtx->session->srctx, RedisGearsPy_RegisterOnDone, bc, &err) != REDISMODULE_OK) {
4831 RedisModule_AbortBlock(bc);
4832 RedisModule_ReplyWithError(ptctx->currentCtx, err);
4833 }
4834 bdiCtx->session->srctx = NULL;
4835 } else {
4836 if(ptctx->createdExecution){
4837 RedisGearsPy_AddSessionRequirementsToDict(bdiCtx->session); // add the requirements to dictionary
4838 bdiCtx->session->fullSerialization = true; // mark the session to be fully serialized with requirements
4839 if(bdiCtx->isBlocking){

Callers 2

RedisGearsPy_ExecuteFunction · 0.85

Calls 6

GetPythonThreadCtxFunction · 0.85
RedisGearsPy_LockFunction · 0.85
getPyErrorFunction · 0.85
RedisGearsPy_UnlockFunction · 0.85
array_lenFunction · 0.85

Tested by

no test coverage detected