Re: [cocci] Searching for duplicate exception handling code with SmPL?

Markus Elfring <[email protected]> Sat, 13 Jun 2026 13:05:59 +0200
Newsgroups fr.inria.cocci
Message-ID <[email protected]>
>     But the desire is growing to reduce the presentation of unwanted
>     false positives considerably.
>     Thus there is a need to switch to more appropriate data formats.
>     It seems then that multiple code occurrences can eventually be taken
>     better into account with the help of SmPL position variables.
>     Their information can be stored in databases probably also according
>     to ACID criteria.

Another SmPL script variant:
@initialize:python@
@@
import sqlalchemy, sys
sys.stderr.write("\n".join(["Using SQLAlchemy version:",
                            sqlalchemy.__version__]))
sys.stderr.write("\n")
from sqlalchemy import Column, Integer, String, create_engine
# outdated: from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.orm import sessionmaker
engine =3D create_engine("sqlite:///:memory:", echo=3DFalse)
# outdated: base =3D declarative_base()

class Base(DeclarativeBase):
   pass

class action(Base):
   __tablename__ =3D "statements"
   name: Mapped[str] =3D mapped_column(String, primary_key=3DTrue)
   source_file: Mapped[str] =3D mapped_column(String, primary_key=3DTrue)
   line: Mapped[int] =3D mapped_column(Integer, primary_key=3DTrue)
   column: Mapped[int] =3D mapped_column(Integer, primary_key=3DTrue)
   value1: Mapped[str] =3D mapped_column(String)
   value2: Mapped[str] =3D mapped_column(String)

   def __repr__(self):
      return """<action(name=3D'%s',
source_file=3D'%s',
line=3D'%s',
column=3D'%s',
value1=3D'%s',
value2=3D'%s')>""" % (self.name,
                              self.source_file,
                              self.line,
                              self.column,
                              self.value1,
                              self.value2)

class action2(Base):
   __tablename__ =3D "statements2"
   name: Mapped[str] =3D mapped_column(String, primary_key=3DTrue)
   source_file: Mapped[str] =3D mapped_column(String, primary_key=3DTrue)
   line: Mapped[int] =3D mapped_column(Integer, primary_key=3DTrue)
   column: Mapped[int] =3D mapped_column(Integer, primary_key=3DTrue)
   value1: Mapped[str] =3D mapped_column(String)
   value2: Mapped[str] =3D mapped_column(String)

   def __repr__(self):
      return """<action2(name=3D'%s',
source_file=3D'%s',
line=3D'%s',
column=3D'%s',
value1=3D'%s',
value2=3D'%s')>""" % (self.name,
                              self.source_file,
                              self.line,
                              self.column,
                              self.value1,
                              self.value2)

configured_session =3D sessionmaker(bind=3Dengine)
session =3D configured_session()
# base.metadata.create_all(engine)
#
# See also:
# https://stackoverflow.com/questions/70402667/how-to-use-create-all-for-s=
qlalchemy-orm-objects-across-files

Base.metadata.create_all(engine)

def store_data(fun, source1, source2, x, y):
    """Add data to internal tables."""
    for place in source1:
       entry =3D action(name =3D fun,
                      source_file =3D place.file,
                      line =3D place.line,
                      column =3D int(place.column) + 1,
                      value1 =3D x,
                      value2 =3D y)
       session.add(entry)

    for place in source2:
       entry =3D action2(name =3D fun,
                       source_file =3D place.file,
                       line =3D place.line,
                       column =3D int(place.column) + 1,
                       value1 =3D x,
                       value2 =3D y)
       session.add(entry)

@searching@
expression e, x;
identifier item, rc, work;
position p1, p2;
type T;
@@
 T work(...)
 {
 ... when any
 if (...)
 {
 ... when !=3D rc =3D e
     when !=3D item =3D x
 kfree@p1(item);
 return rc;
 }
 ... when any
 if (...)
 {
 ... when !=3D rc =3D e
     when !=3D item =3D x
 kfree@p2(item);
 return rc;
 }
 ... when any
 }

@script:python collection@
fun << searching.work;
x << searching.item;
y << searching.rc;
p1 << searching.p1;
p2 << searching.p2;
@@
store_data(fun, p1, p2, x, y)

@finalize:python@
@@
session.commit()
from sqlalchemy import func
entries =3D session.query(func.count("*")).select_from(action).scalar()

if entries > 0:
   from sqlalchemy import Index, Table, MetaData, select, text
   from sqlalchemy.engine.reflection import Inspector
   from sqlalchemy.sql import literal_column

   pairs =3D Index("pairs", action.value1, action.value2)
   pairs.create(engine)

   # See also:
   # https://stackoverflow.com/questions/30575111/how-to-create-a-new-tabl=
e-from-select-statement-in-sqlalchemy#answer-30577608

   q =3D session.query(action.value1, action.value2, action.name, action.s=
ource_file,
                     func.count(literal_column("*")).label("C")
                    ).group_by(action.value1,
                               action.value2,
                               action.name,
                               action.source_file) \
                     .having(func.count(literal_column("*")) > literal_col=
umn("1"))
   session.execute(text('create table t2 as ' + str(q.statement)))

   class results(Base):
       __table__ =3D Table("t2", Base.metadata, autoload_with =3D session.=
connection())
       __mapper_args__ =3D {
           "primary_key": [__table__.c.source_file,
                           __table__.c.name,
                           __table__.c.value1,
                           __table__.c.value2]
                         }

   entries2 =3D session.query(func.count("*")).select_from(results).scalar=
()

   if entries2 > 0:
      delimiter =3D "|"
      sys.stdout.write(delimiter.join(["value1",
                                       "value2",
                                       '"function name"',
                                       '"source file"',
                                       "incidence"]))
      sys.stdout.write("\r\n")
      for value1, value2, name, source_file, incidence \
          in session.query(results.value1,
                           results.value2,
                           results.name,
                           results.source_file,
                           results.C).order_by(results.source_file,
                                               results.name,
                                               results.value1,
                                               results.value2):
         sys.stdout.write(delimiter.join([value1,
                                          value2,
                                          name,
                                          source_file,
                                          str(incidence)]))
         sys.stdout.write("\r\n")
   else:
      sys.stderr.write("Duplicate statements were not determined from "
                       + str(entries) + " records.\n")
      delimiter =3D "|"
      sys.stderr.write(delimiter.join(["value1",
                                       "value2",
                                       '"function name"',
                                       '"source file"',
                                       "line"]))
      sys.stderr.write("\r\n")
      for value1, value2, name, source_file, line \
          in session.query(action.value1,
                           action.value2,
                           action.name,
                           action.source_file,
                           action.line).order_by(action.source_file,
                                               action.name,
                                               action.value1,
                                               action.value2,
                                               action.line):
         sys.stderr.write(delimiter.join([value1,
                                          value2,
                                          name,
                                          source_file,
                                          str(line)]))
         sys.stderr.write("\r\n")
else:
   sys.stderr.write("No result for this analysis!\n")



Corresponding source file example:
https://elixir.bootlin.com/linux/v7.1-rc7/source/sound/core/seq/seq_fifo.c=
#L17-L47


// SPDX-License-Identifier: GPL-2.0-or-later
// deleted part
struct snd_seq_fifo *snd_seq_fifo_new(int poolsize)
{
	struct snd_seq_fifo *f;

	f =3D kzalloc_obj(*f);
	if (!f)
		return NULL;

	f->pool =3D snd_seq_pool_new(poolsize);
	if (f->pool =3D=3D NULL) {
		kfree(f);
		return NULL;
	}
	if (snd_seq_pool_init(f->pool) < 0) {
		snd_seq_pool_delete(&f->pool);
		kfree(f);
		return NULL;
	}
// deleted part
	return f;
}
// deleted part



Questionable test results:
Markus_Elfring@Sonne:=E2=80=A6/Projekte/Linux/next-analyses> git checkout =
next-20260608 && time /usr/bin/spatch =E2=80=A6/Projekte/Coccinelle/janito=
r/list_selected_duplicate_statements_in_if_branches.cocci sound/core/seq/s=
eq_fifo.c
=E2=80=A6
Using SQLAlchemy version:
2.0.49
=E2=80=A6
Duplicate statements were not determined from 1 records.
value1|value2|"function name"|"source file"|line
f|NULL|snd_seq_fifo_new|sound/core/seq/seq_fifo.c|28

real    0m0,816s
user    0m0,658s
sys     0m0,148s

Markus_Elfring@Sonne:=E2=80=A6/Projekte/Coccinelle/Probe> time /usr/bin/sp=
atch ../janitor/list_selected_duplicate_statements_in_if_branches.cocci sn=
d_seq_fifo_new-excerpt-20260608.c
=E2=80=A6
Using SQLAlchemy version:
2.0.49
=E2=80=A6
Duplicate statements were not determined from 1 records.
value1|value2|"function name"|"source file"|line
f|NULL|snd_seq_fifo_new|snd_seq_fifo_new-excerpt-20260608.c|13

real    0m0,785s
user    0m0,649s
sys     0m0,128s



Would anybody like to explain the different determined values?

How can presented expertise challenges be resolved?

Regards,
Markus