如何从子进程写入Flask Session及正确返回字符串？

问题背景 #

开发了一个图像处理Flask Web应用，父进程调用的函数会生成多张临时图片并写入磁盘，同时将图片路径存入Flask Session。但该流程存在内存泄漏，占用的内存必须重启服务器才能释放。尝试用独立子进程调用该函数时，遇到两个问题：

• 子进程中添加到Session的键值对无法同步到父进程；
• 子进程写入multiprocessing.Value的字符串无法同步到父进程。

复现代码（app.py） #

from flask import Flask, session
from flask_session import Session
from multiprocessing import Process, Value

app = Flask(__name__.split('.')[0])
app.secret_key = "secret_key"

SESSION_TYPE = 'filesystem'
app.config.from_object(__name__)
Session(app)


def write_to_session(session):
    session['inside_function'] = "inside"
    print(session.items())

# 基准测试：函数直接修改Session（预期行为）
@app.route('/baseline', methods=['GET'])
def baseline():
    session.clear()
    session['outside_function'] = "outside"

    text = ""
    text += "<p>Initial session data <br />" + str(session.items()) + "</p>"
    write_to_session(session)
    text += "<p>Session after function call <br />" + str(session.items()) + "</p>"

    return text

# 子进程修改Session：子进程内修改有效，但父进程看不到变化
@app.route('/multi', methods=['GET'])
def multi():
    session.clear()
    session['outside_function'] = "outside"

    text = ""
    text += "<p>Initial session data <br />" + str(session.items()) + "</p>"

    process = Process(target=write_to_session, args=(session,))
    process.start()
    process.join() # 等待子进程结束

    text += "<p>Session after function call <br />" + str(session.items()) + "</p>"

    return text

# 用multiprocessing.Value传递字符串：子进程赋值后父进程拿到的是无效值
@app.route('/value', methods=['GET'])
def value():
    from ctypes import c_wchar_p

    def assign_string(cstring):
        cstring.value = "inside"

    cstring = Value(c_wchar_p, "Hello initial!")
    text = "<p>Initial Value:<br />" + str(cstring.value) + "</p>"

    process = Process(target=assign_string, args=(cstring,))
    process.start()
    process.join() 

    text += "<p>Value after function call:<br />" + str(cstring.value) + "</p>"

    return text


if __name__ == '__main__':
    app.run(host='localhost', port=5000, debug=True, use_reloader=True)

测试结果 #

预期行为（/baseline） #

访问http://localhost:5000/baseline返回：

Initial session data
dict_items([('outside_function', 'outside')])
Session after function call
dict_items([('outside_function', 'outside'), ('inside_function', 'inside')])

子进程写入Session失败（/multi） #

访问http://localhost:5000/multi返回：

Initial session data
dict_items([('outside_function', 'outside')])
Session after function call
dict_items([('outside_function', 'outside')])

multiprocessing.Value传递字符串失败（/value） #

访问http://localhost:5000/value返回：

Initial Value:
Hello initial!
Value after function call:

Anaconda环境配置 #

name: flask
channels:
  - conda-forge
  - defaults
dependencies:
  - _libgcc_mutex=0.1=conda_forge
  - _openmp_mutex=4.5=2_gnu
  - bzip2=1.0.8=h4bc722e_7
  - ca-certificates=2024.8.30=hbcca054_0
  - cachelib=0.13.0=pyhd8ed1ab_0
  - click=8.1.7=unix_pyh707e725_0
  - flask=2.2.0=pyhd8ed1ab_0
  - flask-session=0.5.0=pyhd8ed1ab_0
  - importlib-metadata=8.4.0=pyha770c72_0
  - itsdangerous=2.2.0=pyhd8ed1ab_0
  - jinja2=3.1.4=pyhd8ed1ab_0
  - ld_impl_linux-64=2.40=hf3520f5_7
  - libexpat=2.6.3=h5888daf_0
  - libffi=3.4.2=h7f98852_5
  - libgcc=14.1.0=h77fa898_1
  - libgcc-ng=14.1.0=h69a702a_1
  - libgomp=14.1.0=h77fa898_1
  - libnsl=2.0.1=hd590300_0
  - libsqlite=3.46.1=hadc24fc_0
  - libuuid=2.38.1=h0b41bf4_0
  - libxcrypt=4.4.36=hd590300_1
  - libzlib=1.3.1=h4ab18f5_1
  - markupsafe=2.1.5=py311h9ecbd09_1
  - ncurses=6.5=he02047a_1
  - openssl=3.3.2=hb9d3cd8_0
  - pip=24.2=pyh8b19718_1
  - python=3.11.0=he550d4f_1_cpython
  - python_abi=3.11=5_cp311
  - readline=8.2=h8228510_1
  - setuptools=73.0.1=pyhd8ed1ab_0
  - tk=8.6.13=noxft_h4845f30_101
  - tzdata=2024a=h8827d51_1
  - werkzeug=2.2.2=pyhd8ed1ab_0
  - wheel=0.44.0=pyhd8ed1ab_0
  - xz=5.2.6=h166bdaf_0
  - zipp=3.20.1=pyhd8ed1ab_0
prefix: /home/rh/.local/share/mambaforge-pypy3/envs/flask

问题原因与解决方案 #

1. 子进程写入Flask Session的正确方式 #

原因：多进程模式下，父进程的Session对象会被复制到子进程中，子进程修改的是副本而非父进程原对象。同时子进程没有请求上下文，无法正确触发Session的持久化逻辑，导致修改无法同步。

解决方案：
所有Session修改操作由父进程完成，子进程仅处理业务逻辑并将需要存入Session的数据返回给父进程，推荐用multiprocessing.Queue传递数据：

from multiprocessing import Queue

def process_image_task(queue):
    # 模拟图像处理生成路径
    image_paths = ["tmp/img1.jpg", "tmp/img2.jpg"]
    queue.put(image_paths)

@app.route('/process_image', methods=['GET'])
def process_image():
    session.clear()
    queue = Queue()
    p = Process(target=process_image_task, args=(queue,))
    p.start()
    p.join()
    # 父进程获取数据并写入Session
    image_paths = queue.get()
    session['image_paths'] = image_paths
    return str(session.items())

2. 子进程返回字符串的正确方式 #

原因：c_wchar_p是内存指针，子进程赋值的字符串内存地址仅在子进程有效，父进程无法访问该内存空间，导致拿到无效值。

解决方案：

• 优先使用multiprocessing.Queue传递字符串，这是跨进程数据传递的安全方式：

  from multiprocessing import Queue
  
  def get_result(queue):
      queue.put("inside")
  
  @app.route('/value_fix', methods=['GET'])
  def value_fix():
      queue = Queue()
      process = Process(target=get_result, args=(queue,))
      process.start()
      process.join()
      result = queue.get()
      return f"<p>Value after function call:<br />{result}</p>"
  

• 若必须使用共享内存，可改用固定长度的字符数组（如Array），但需预先确定字符串长度：

  from multiprocessing import Array
  from ctypes import c_char
  
  def assign_string(arr):
      arr.value = b"inside"
  
  @app.route('/value_fix_array', methods=['GET'])
  def value_fix_array():
      # 初始化长度为10的字符数组
      arr = Array(c_char, 10)
      arr.value = b"Hello initial!"
      text = f"<p>Initial Value:<br />{arr.value.decode()}</p>"
      process = Process(target=assign_string, args=(arr,))
      process.start()
      process.join()
      text += f"<p>Value after function call:<br />{arr.value.decode()}</p>"
      return text
  

内容的提问来源于stack exchange，提问作者RedHand